pyspark 中的时间戳验证答案

【问题标题】：Timestamp validation in pysparkpyspark 中的时间戳验证
【发布时间】：2020-08-29 21:21:29
【问题描述】：

我们正在 pyspark 中构建数据摄取框架并尝试处理时间戳异常。基本上，希望在单独的列中有一个拒绝记录，该记录不与架构确认。

df = spark.createDataFrame(
    [
        ("1988-06-15 11:55:12.1","1"),
        ("1988-06-14 11:55:12", "3"),
        ("1988-06-13 11:55:12","1"),
        ("1988-06-12 11:55:12", "2")
    ],
    ['timestampColm','intColm']
)

在名为 badRecords 的数据框中创建一个新列，以捕获此数据框中可能存在的所有错误，并使用“yyyy-MM-dd HH:mm:dd”格式验证时间戳列。

尝试使用以下代码验证时间戳

示例 1

df1 = df.withColumn("badRecords",
                f.when(
                        to_timestamp(f.col("timestampColm"), "yyyy-MM-dd HH:mm:ss").cast("Timestamp").isNull() & f.col("timestampColm").isNotNull(),f.lit("Not a valid Timestamp")
                       ).otherwise(f.lit(None))
              )

因此它应该将第一条记录“1988-06-15 11:55:12.1”标记为无效，因为它不支持“HH:mm:ss”格式，但它仍在验证记录而不是拒绝它。

+--------------------+-----------+----------+
|       timestampColm|    intColm|badRecords|
+--------------------+-----------+----------+
|1988-06-15 11:55:...|          1|      null|
| 1988-06-14 11:55:12|       null|      null|
| 1988-06-13 11:55:12|          1|      null|
| 1988-06-12 11:55:12|          2|      null|
+--------------------+-----------+----------+

经过几次分析发现我们可以用 unix_timestamp 做到这一点，但没有运气

示例 2

df1 = df.withColumn("badRecords",
                      f.when(
                            f.from_unixtime(
                                  f.unix_timestamp(
                                         f.col("timestampColm"),"yyyy-MM-dd HH:mm:ss")
                            ).cast("timestamp").isNull() & f.col("timestampColm").isNotNull(),
                            f.lit("Not a valid Timestamp")
                    ).otherwise(f.lit(None))
                )

帮助我了解我缺少什么，因为它仍在验证而不是拒绝记录？

【问题讨论】：

标签： pyspark apache-spark-sql pyspark-dataframes to-timestamp

【解决方案1】：

在您的情况下，您已经编写了 & 但它应该是“和”以便进行逻辑运算。 '&' 是位运算符。可能在 pyspark 中被视为逻辑运算符。考虑尝试这个-： df1 = df.withColumn("badRecords", f.当（ (to_timestamp(f.col("timestampColm"), "yyyy-MM-dd HH:mm:ss").cast("Timestamp").isNull()) & (f.col("timestampColm").isNotNull( )),f.lit("不是一个有效的时间戳") ).otherwise(f.lit(无)) )

我的意思是考虑添加括号并将每个条件括起来，例如 (condition1) & (condition2)。希望这会有所帮助。

【讨论】：

感谢您的回复，但面临 ValueError: Cannot convert column into bool: please use '&' for 'and', '|'在构建 DataFrame 布尔表达式时，为 'or'，'~' 为 'not'。
请尝试修改后的方法，看看是否有帮助。干杯！
运气不好！问题在于 f.unix_timestamp(f.col("dateAsString"),"yyyy-MM-dd HH:mm:ss") ** 它以秒为单位转换值，因此它不考虑毫秒/NanoSconds 因此字符串我通过 **1988-06-15 11:55:12.1 它仅以 (HH:mm:ss) 格式读取它，因此验证而不是拒绝
你可以尝试使用自定义 udf 来转换时间戳。我已经做到了，而且效果很好。

【解决方案2】：

我可以通过创建自定义 UDF 来解决这个问题，它工作正常。

validate_timestamp_udf = udf(lambda val: validate_timestamp(val))
df6 = df2.withColumn("badRecords",validate_timestamp_udf(f.col(ColName)))

并且在 validate_timestamp() 函数中，我在正则表达式的帮助下进行格式验证。

【讨论】：