Scala Spark SQL高效实现：加载ID文件作为WHERE子句过滤条件

问题场景 #

我有一个每行存储一个BigInteger类型ID的TXT文件（行数超过10万），需要用Scala编写Spark SQL查询，读取这些ID并作为WHERE子句的过滤条件，只查询对应ID的数据。原本考虑把ID存入集合来判断存在性，但不确定这种场景下的最优高效方案。

示例TXT内容：

1234566789
9876543212

原本的思路（存在性能隐患）：

// 假设已将ID读取到集合idSet中
spark.sql(f"""
  SELECT
    table_x.id,
    AVG(table_x.cost)
  FROM table_x
  WHERE table_x.id IN (${idSet.mkString(",")})
  GROUP BY table_x.id
  """)

最优实现方案 #

直接用集合生成IN语句或存集合判断存在性，在ID数量过万后会出现两个核心问题：一是生成的SQL语句过长，解析成本极高；二是大集合广播会占用过多Executor内存，拖慢整体任务。最优方案是将TXT文件读成Spark DataFrame，通过JOIN或子查询实现过滤，Spark会自动优化执行计划（比如广播小表、分区裁剪），性能远优于集合过滤。

方式一：DataFrame JOIN实现

把TXT文件读成临时视图，和目标表做JOIN后再聚合：

// 1. 读取TXT文件为DataFrame，根据目标表ID类型调整转换规则（超大数用decimal(20,0)）
val idDF = spark.read.textFile("path/to/your/id_file.txt")
  .toDF("filter_id")
  .select($"filter_id".cast("long").alias("id"))

// 2. 创建临时视图供SQL调用
idDF.createOrReplaceTempView("filter_ids")

// 3. 执行JOIN查询
val resultDF = spark.sql("""
  SELECT
    table_x.id,
    AVG(table_x.cost) AS avg_cost
  FROM table_x
  JOIN filter_ids ON table_x.id = filter_ids.id
  GROUP BY table_x.id
""")

resultDF.show()

方式二：子查询IN实现

如果更习惯WHERE IN的写法，可将ID表作为子查询，Spark会自动优化执行逻辑：

// 1. 读取TXT文件并创建临时视图（同方式一）
val idDF = spark.read.textFile("path/to/your/id_file.txt")
  .toDF("filter_id")
  .select($"filter_id".cast("long").alias("id"))
idDF.createOrReplaceTempView("filter_ids")

// 2. 执行子查询过滤
val resultDF = spark.sql("""
  SELECT
    table_x.id,
    AVG(table_x.cost) AS avg_cost
  FROM table_x
  WHERE table_x.id IN (SELECT id FROM filter_ids)
  GROUP BY table_x.id
""")

resultDF.show()

关键注意事项 #

• 类型严格匹配：必须保证读取的ID类型和table_x.id完全一致（比如都是Long或Decimal），否则隐式类型转换会大幅降低过滤/JOIN性能。
• 广播阈值调整：如果ID表大小接近或超过默认广播阈值（10MB），可通过spark.sql.autoBroadcastJoinThreshold参数调整，强制Spark广播ID表，避免Shuffle操作。
• 禁止硬编码ID集合：绝对不要把10万级ID拼接成IN语句，这会导致SQL解析超时，甚至超出语句长度限制。

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