Spark内连接返回空数据集问题：大文件场景异常排查

你遇到的这个问题挺典型——小文件时内连接能正常返回结果，换成20GB的大文件就直接空集，但用filter+isin却能正常拿到匹配数据，而且两个表的DAY_ID字段都是string类型。结合Spark的执行机制，我梳理了几个最可能的原因：

1. Join策略切换引发的匹配异常 #

Spark会根据数据量自动选择最优Join策略：

• 处理小文件时，days_df因为数据量极小，Spark会默认采用Broadcast Hash Join：把小表广播到所有Executor节点，和大表的每一行做本地精确匹配，这种模式下不容易出现匹配失效的问题。
• 但处理大文件时，Spark会自动切换到Shuffle Hash Join或Sort Merge Join：这时候需要对两个表的DAY_ID做哈希分区，再通过Shuffle传输数据到对应分区进行匹配。如果哈希计算过程中出现隐性不一致（比如字符串编码差异、不可见字符导致哈希值不同），就会出现明明存在匹配值，但分区后无法对应的情况。而filter+isin不需要Shuffle，直接在每个分区本地做匹配，自然能正常命中数据。

2. 大文件中的隐性脏数据干扰 #

虽然两个表的DAY_ID都是string类型，但大文件里可能存在小文件没有的隐性脏数据：比如部分DAY_ID带有不可见字符（空格、制表符、换行符）、大小写不一致，或者存在前导/后置空格。不过这里有个矛盾——如果是这类脏数据，filter+isin也应该匹配不上，除非你的days_array本身就是从大文件中提取的（连带脏数据），但Spark读取大文件时部分分区的DAY_ID解析出现了格式偏差，而filter+isin的本地匹配逻辑刚好能兼容这种偏差？

3. 代码的隐性语法问题 #

看你给出的Join代码：

val filtered_sales = sales.join(days_df,Seq("DAY_ID")

这里少了一个右括号，正确写法应该是sales.join(days_df, Seq("DAY_ID"))。不过你说小文件场景下能正常运行，大概率是输入时的笔误，但如果实际运行时确实存在语法问题，大文件场景下Spark的容错机制可能直接返回空集而非报错，这个可能性虽然小，但还是建议先确认代码的正确性。

解决方案建议 #

针对这些可能的原因，你可以试试以下方法：

• 强制使用Broadcast Join：手动指定广播小表，避免Shuffle带来的匹配问题：

  import org.apache.spark.sql.functions.broadcast
  val filtered_sales = sales.join(broadcast(days_df), Seq("DAY_ID"))
  filtered_sales.show()
  

• 清洗DAY_ID字段：对两个表的DAY_ID做统一清洗，确保格式完全一致：

  import org.apache.spark.sql.functions.{trim, col}
  val cleaned_sales = sales.withColumn("DAY_ID", trim(col("DAY_ID")))
  val cleaned_days_df = days_df.withColumn("DAY_ID", trim(col("DAY_ID")))
  val filtered_sales = cleaned_sales.join(cleaned_days_df, Seq("DAY_ID"))
  

• 排查Shuffle日志：检查Spark的Shuffle相关日志，看是否有数据丢失、分区异常的报错信息，这能帮你定位具体的执行问题。

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