我们正在 pyspark 中构建数据摄取框架,并想知道处理数据类型异常的最佳方法是什么。基本上,我们希望有一个拒绝表来捕获所有未与架构确认的数据。
stringDf = sparkSession.createDataFrame(
[
("11/25/1991","1"),
("11/24/1991", None),
("11/30/1991","a")
],
['dateAsString','intAsString']
)
这是我的 stringDf
有两列。
+------------+-----------+
|dateAsString|intAsString|
+------------+-----------+
| 11/25/1991| 1|
| 11/24/1991| null|
| 11/30/1991| a|
+------------+-----------+
我想为数据框创建一个名为 dataTypeValidationErrors
的新列,以捕获此数据集中可能存在的所有错误。使用 pyspark 实现这一目标的最佳方法是什么?
+------------+-----------+------------------------+
|dateAsString|intAsString|dataTypeValidationErrors|
+------------+-----------+------------------------+
| 11/25/1991| 1|None |
| 11/24/1991| null|None |
| 11/30/1991| a|Not a valid Number |
+------------+-----------+------------------------+
最佳答案
您可以尝试将列 cast
为所需的 DataType。如果存在不匹配或错误,将返回 null
。在这些情况下,您需要验证原始值不是 null
,如果不是,则存在错误。
- 使用
pyspark.sql.functions.when()
测试转换后的列是否为null
且原始值不是null
。 - 如果这是
True
,则使用字符串文字"Not a valid Number"
作为列值。否则返回字符串"None"
。
例如:
import pyspark.sql.functions as f
stringDf.withColumn(
"dataTypeValidationErrors",
f.when(
f.col("intAsString").cast("int").isNull() & f.col("intAsString").isNotNull(),
f.lit("Not a valid Number")
).otherwise(f.lit("None"))
)\
.show()
#+------------+-----------+------------------------+
#|dateAsString|intAsString|dataTypeValidationErrors|
#+------------+-----------+------------------------+
#| 11/25/1991| 1| None|
#| 11/24/1991| null| None|
#| 11/30/1991| a| Not a valid Number|
#+------------+-----------+------------------------+
您也可以将其扩展到多列:
假设您还有一行包含无效的 dateAsString
值:
stringDf = spark.createDataFrame(
[
("11/25/1991","1"),
("11/24/1991", None),
("11/30/1991","a"),
("13.14.15", "b")
],
['dateAsString','intAsString']
)
使用字典定义每一列的转换:
conversions = {
'dateAsString':lambda c: f.from_unixtime(f.unix_timestamp(c,"MM/dd/yyyy")).cast("date"),
'intAsString':lambda c: f.col(c).cast('int')
}
stringDf.withColumn(
"dataTypeValidationErrors",
f.concat_ws(", ",
*[
f.when(
v(k).isNull() & f.col(k).isNotNull(),
f.lit(k + " not valid")
).otherwise(f.lit(None))
for k, v in conversions.items()
]
)
)\
.show(truncate=False)
#+------------+-----------+---------------------------------------------+
#|dateAsString|intAsString|dataTypeValidationErrors |
#+------------+-----------+---------------------------------------------+
#|11/25/1991 |1 | |
#|11/24/1991 |null | |
#|11/30/1991 |a |intAsString not valid |
#|13.14.15 |b |dateAsString not valid, intAsString not valid|
#+------------+-----------+---------------------------------------------+
或者如果你只是想知道一行是否有错误,而不需要知道细节:
stringDf.withColumn(
"dataTypeValidationErrors",
f.when(
reduce(
lambda a, b: a|b,
(v(k).isNull() & f.col(k).isNotNull() for k, v in conversions.items())
),
f.lit("Validation Error")
).otherwise(f.lit("None"))
)\
.show(truncate=False)
#+------------+-----------+------------------------+
#|dateAsString|intAsString|dataTypeValidationErrors|
#+------------+-----------+------------------------+
#|11/25/1991 |1 |None |
#|11/24/1991 |null |None |
#|11/30/1991 |a |Validation Error |
#|13.14.15 |b |Validation Error |
#+------------+-----------+------------------------+
关于pyspark - pyspark 中的数据类型验证,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51881686/