如何使用PySpark从Spark DataFrame中筛选并拆分有效/无效时间戳记录

实现思路 #

利用PySpark内置的to_timestamp函数对timestamp列做格式转换，转换失败会返回null，以此为依据拆分有效和无效记录即可。

代码实现 #

1. 导入依赖函数 #

from pyspark.sql.functions import to_timestamp, col

2. 标记时间戳是否合法 #

如果你的时间戳有特定格式，需要在to_timestamp的第二个参数传入对应格式字符串，比如常见的yyyy-MM-dd HH:mm:ss，否则函数会用默认规则尝试转换。

# 假设原始DataFrame名为source_df
parsed_df = source_df.withColumn(
    "is_valid_ts", 
    to_timestamp(col("timestamp")).isNotNull()
    # 带格式的写法示例：to_timestamp(col("timestamp"), "yyyy-MM-dd HH:mm:ss").isNotNull()
)

3. 拆分两个子DataFrame #

# 仅保留时间戳合法的记录
valid_df = parsed_df.filter(col("is_valid_ts") == True).drop("is_valid_ts")
# 仅保留时间戳非法的记录
invalid_df = parsed_df.filter(col("is_valid_ts") == False).drop("is_valid_ts")

4. 结果验证 #

# 查看有效数据
valid_df.show()
# 查看无效记录
invalid_df.show()

注意事项 #

• 如果你的timestamp列存储的是毫秒级时间戳数字，转换时需要先除以1000再传入to_timestamp，示例：to_timestamp(col("timestamp") / 1000).isNotNull()
• 格式字符串要和你的实际时间戳文本格式完全匹配，避免合法时间戳被误判为无效。

内容的提问来源于stack exchange，提问作者AshrithGande