Databricks中Join操作仅返回部分结果的问题排查求助

1. 关联键数据类型不匹配 #

Parquet是强类型存储，CSV导入时自动推断类型极易出错。比如Parquet的关联键是long类型，CSV被推断为string，表面值一致但类型不匹配会导致大量匹配失败。

• 排查：打印两个DataFrame的Schema确认类型：

  parquet_df.printSchema()
  csv_df.printSchema()
  

• 修复：手动对齐关联键类型，例如将CSV的键转为long：

  from pyspark.sql.types import LongType
  csv_df = csv_df.withColumn("join_key", csv_df["join_key"].cast(LongType()))
  

2. 空值/不可见字符导致匹配失效 #

Excel、Alteryx对空值或不可见字符的处理逻辑和Spark不同：Spark中空值无法互相匹配，而其他工具可能自动忽略；另外CSV中常混入空格、换行符等不可见字符，肉眼无法察觉但会导致匹配失败。

• 排查空值数量：

  parquet_df.filter(parquet_df.join_key.isNull()).count()
  csv_df.filter(csv_df.join_key.isNull()).count()
  

• 清理不可见字符：

  from pyspark.sql.functions import trim, regexp_replace
  # 去除前后空格及换行符
  csv_df = csv_df.withColumn("join_key", regexp_replace(trim(csv_df.join_key), "\n|\r", ""))
  parquet_df = parquet_df.withColumn("join_key", regexp_replace(trim(parquet_df.join_key), "\n|\r", ""))
  

3. 任务跳过背后的数据读取问题 #

日志中的「跳过任务」通常是Spark自动跳过空分区/空文件，但如果部分有效数据未被读取，也会导致结果行数减少。

• 验证数据完整性：对比CSV文件总行数与DataFrame计数，检查Parquet是否读取全部分区：

  # 确认CSV导入行数
  csv_df.count()
  # 查看Parquet分区数据分布
  parquet_df.rdd.mapPartitions(lambda x: [sum(1 for _ in x)]).collect()
  

• 查看Spark UI：进入对应Job的Stage页面，检查每个Stage的输入数据量是否符合预期，确认是否有数据未被加载。

4. Join类型与其他工具不一致 #

Excel的VLOOKUP默认是左匹配（等价于Spark的left join），如果在Databricks中误用inner join，结果行数会大幅减少；反之亦然。

• 检查SQL语句中的Join类型，确保与其他工具逻辑一致：

  -- 例如和VLOOKUP逻辑对齐的左连接
  SELECT * FROM parquet_df LEFT JOIN csv_df ON parquet_df.join_key = csv_df.join_key
  

5. 广播Join未生效或Shuffle异常 #

虽然指定了广播Join，但如果广播的DataFrame体积超过spark.sql.autoBroadcastJoinThreshold（默认10MB），Spark会自动退化为Shuffle Join，极端情况下可能出现数据丢失。

• 强制广播并验证：

  from pyspark.sql.functions import broadcast
  # 强制广播小表（CSV通常体积更小）
  result_df = parquet_df.join(broadcast(csv_df), on="join_key", how="left")
  result_df.count()
  

• 检查Shuffle配置：确认spark.shuffle.service.enabled为true（Databricks默认开启），避免Shuffle过程中数据丢失。

6. Parquet文件损坏或路径错误 #

Parquet文件可能存在损坏，或者读取路径仅包含部分分区/文件，导致只加载了20%的数据。

• 检查Parquet存储路径：

  dbutils.fs.ls("/path/to/your/parquet/data")
  

• 验证Parquet总行数：对比源数据的总行数与parquet_df.count()的结果，确认是否完整读取。

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