PySpark基于多列组合为键补全缺失数据及连接异常排查

你最初的思路是可行的，更简洁的写法不需要提前给全量组合DF加value列，左连接后用coalesce函数处理空值即可，代码如下：

# 1. 生成全量合法组合的骨架DF，无需提前设置默认value
skeleton_df = TS_DF.crossJoin(COL1_DF).crossJoin(COL2_DF)
# 2. 左连接部分数据，三个列名一致时直接传列表作为匹配键即可
merged_df = skeleton_df.join(
    partial_data_df,
    on=["ts", "col1", "col2"],
    how="left"
)
# 3. 未匹配到的value空值替换为0
merged_df = merged_df.withColumn("value", F.coalesce(F.col("value"), F.lit(0)))

结合你给出的代码和统计数据，丢失匹配的核心原因大概率是以下几个：

• 连接条件列名不匹配：你生成EMPTY_DF时的列名是ts/col1/col2，但连接条件里写的是EMPTY_DF.ets/ecol1/ecol2，如果你没有额外给骨架DF重命名列，这几个列实际不存在，Spark会判定匹配条件恒假，导致大量partial_data_df的数据匹配失败，和你统计的非0值大量丢失的现象吻合。
• 时间戳精度不匹配：你生成TS_DF时的步长是1小时（1000*3600），所有ts都是整小时时间戳，如果你读取的partial_data_df中的ts不是整小时（包含分钟、秒偏移），就会出现匹配失败。可以先对partial_data_df的ts做整小时对齐再连接：

  partial_data_df = partial_data_df.withColumn("ts", F.floor(F.col("ts") / 3600000) * 3600000)
  

• 数据类型不匹配：读csv时如果没有显式指定schema，Spark自动推断类型可能把ts列识别为字符串类型，和骨架DF中long类型的ts做等值判断时隐式转换失败，导致匹配不上。建议读取csv时显式指定schema：

  from pyspark.sql.types import StructType, StructField, LongType, StringType
  schema = StructType([
      StructField("ts", LongType(), True),
      StructField("col1", StringType(), True),
      StructField("col2", StringType(), True),
      StructField("value", LongType(), True)
  ])
  partial_data_df = spark.read.schema(schema).csv(my_path)
  

• 枚举值不匹配：你生成COL1_DF、COL2_DF用的枚举值，和partial_data_df中实际的col1、col2值存在差异，比如大小写差异、前后空格、特殊字符转义问题，也会导致匹配失败。

内容的提问来源于stack exchange，提问作者Mr T.