数据湖查询比Hive还慢？——谓词下推和向量化执行怎么配置？

// 查看执行计划（发现问题）
val df = spark.sql(""
  SELECT * FROM user_profile
  WHERE dt='2024-03-19' AND age > 25
""")
df.explain(True)  // 查看详细执行计划
 
// Hive表的执行计划：
// * Filter (age > 25)  -- 谓词下推
//   * Scan HiveTable  -- PartitionFilters: [dt=2024-03-19]
 
// Delta Lake表的执行计划（问题）：
// * Filter (dt=2024-03-19 AND age > 25)  -- 没有谓词下推！
//   * Scan DeltaTable  -- 没有PartitionFilters！

// 谓词下推示例
val df = spark.read.parquet("user_profile")
  .filter("age > 25")  // 过滤条件
  .select("user_id", "age", "gender")
 
// Parquet的执行过程：
// 1) 读取Parquet文件的Min/Max统计信息
// 2) 跳过age.Max <= 25的Row Group（不读数据）
// 3) 只读age.Max > 25的Row Group的数据
// 4) 只读3列（user_id、age、gender）
 
// 如果谓词下推失效：
// 1) 读取所有Row Group的所有数据
// 2) 在内存中过滤age > 25
// 3) IO和内存消耗都大很多

// 谓词下推失效的例子
// 情况1：使用UDF
spark.udf.register("isAdult", (age: Int) => age >= 18)
val df = spark.sql("SELECT * FROM user_profile WHERE isAdult(age)")
// 谓词下推失效（UDF无法下推到Parquet）
 
// 情况2：子查询
val df = spark.sql("""
  SELECT * FROM user_profile
  WHERE age > (SELECT AVG(age) FROM user_profile WHERE country='US')
""")
// 谓词下推失效（子查询无法下推）
 
// 正确做法：改写SQL
val avgAge = spark.sql("SELECT AVG(age) FROM user_profile WHERE country='US'").first().getInt(0)
val df = spark.sql(s"SELECT * FROM user_profile WHERE age > $avgAge")

向量化执行：配置和验证

向量化执行（Vectorized Execution）是'批量处理数据'的技术，一次处理多行（如100行），而不是一次处理一行。这样可以减少函数调用开销，提升CPU效率。

// 向量化执行示例（Parquet）
// 默认开启（spark.sql.parquet.enableVectorizedReader=true）
val df = spark.read.parquet("user_profile")
 
// 非向量化执行：一次处理一行
// for (row <- df.collect()) {
//   process(row)  // 函数调用次数 = 行数
// }
 
// 向量化执行：一次处理100行
// for (batch <- df.collect().grouped(100)) {
//   process(batch)  // 函数调用次数 = 行数 / 100
// }

配置参数：Spark默认开启向量化执行（spark.sql.parquet.enableVectorizedReader=true），但需要注意：1）向量化执行只支持Parquet/ORC，不支持Avro/CSV/JSON；2）如果数据包含'复杂类型'（如Map、Array），向量化执行可能失效；3）向量化执行的'批次大小'可以通过spark.sql.parquet.vectorized.reader.batchSize调整（默认128行）。

// 向量化执行配置
// 1) 启用向量化执行（默认开启）
spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true")
 
// 2) 调整批次大小（默认128行）
spark.conf.set("spark.sql.parquet.vectorized.reader.batchSize", "256")
 
// 3) ORC向量化执行
spark.conf.set("spark.sql.orc.enableVectorizedReader", "true")
spark.conf.set("spark.sql.orc.vectorized.reader.batchSize", "256")

实战效果：我们在出海硬件品牌测试过：'查询大表（10亿行）的5列'，启用向量化执行后，查询延迟从3分钟降到2分钟（提升33%）。

统计信息：为什么重要？

统计信息（Statistics）是'CBO（Cost-Based Optimizer）'的基础，Spark根据统计信息'选择最优执行计划'。如果统计信息过期或缺失，Spark可能选择'差的执行计划'，导致查询慢。

-- 收集统计信息
ANALYZE TABLE user_profile COMPUTE STATISTICS;
 
-- 收集列级统计信息（更详细）
ANALYZE TABLE user_profile COMPUTE STATISTICS FOR COLUMNS user_id, age, gender;
 
-- 查看统计信息
DESCRIBE EXTENDED user_profile;

统计信息包含：1）表级：行数、文件数、大小；2）列级：唯一值数量（NDV）、Min/Max、Null数量、平均长度、直方图。

实战经验：我们在出海硬件品牌遇到过：'查询WHERE age > 25'，Spark错误地选择了'Broadcast Join'（因为统计信息显示小表），但实际是'大表Join'，导致Driver OOM。重新收集统计信息后，Spark选择了'SortMergeJoin'，问题解决。

性能调优参数清单

参数	默认值	推荐值	说明
spark.sql.optimizer.predicatePushdown.enabled	true	true	启用谓词下推
spark.sql.parquet.enableVectorizedReader	true	true	启用Parquet向量化执行
spark.sql.parquet.vectorized.reader.batchSize	128	256	向量化批次大小
spark.sql.orc.enableVectorizedReader	true	true	启用ORC向量化执行
spark.sql.statistics.histogram.enabled	false	true	启用直方图（更精准统计）
spark.sql.autoBroadcastJoinThreshold	10MB	200MB	Broadcast Join阈值
spark.sql.adaptive.enabled	false	true	启用AQE（自适应查询执行）
spark.sql.parquet.filterPushdown	true	true	Parquet谓词下推

面试高频追问

追问一：谓词下推失效了怎么排查？

用'df.explain(True)'查看执行计划，如果没有看到'PushedFilters'或'PartitionFilters'，说明谓词下推失效。常见原因：1）使用UDF（改用内置函数）；2）过滤条件包含子查询（改写SQL）；3）数据源不支持谓词下推（换Parquet/ORC）。

追问二：向量化执行一定比非向量化快吗？

大部分情况是，但不是绝对。向量化执行要求'数据格式简单'（没有复杂类型），如果数据包含大量Map/Array/Struct，向量化执行可能失效，反而不如非向量化。另外，向量化执行的'批次大小'需要调优，太大可能导致OOM，太小无法发挥优势。

追问三：统计信息多久更新一次？

建议'每次有大变动后更新'，如：1）每天导入新数据后；2）表结构变更后；3）发现查询性能突然下降后。可以用'ANALYZE TABLE ... COMPUTE STATISTICS'手动收集，也可以配置'自动收集'（spark.sql.statistics.autoSize.enabled=true）。

实战踩坑：统计信息过期导致查询变慢

2023年在某出海硬件品牌，我们遇到过一次'统计信息过期导致查询变慢'的问题。当时我们的'订单表（orders）'每天都有新数据，但忘记更新统计信息。

运行2个月后，业务部门反馈'订单报表查询越来越慢'——从最初的10秒变成了3分钟。

// 问题查询：Join订单表和用户表
val df = spark.sql("""
  SELECT o.*, u.user_name
  FROM orders o
  JOIN users u ON o.user_id = u.user_id
  WHERE o.dt='2024-03-19'
""")
 
// 执行计划显示：Spark选择了Broadcast Join
// 但orders表已经是'大表'（2亿行），不应该Broadcast
// 结果：Driver OOM

排查发现：1）统计信息显示orders表有'100万行'（2个月前的数据）；2）实际orders表已经有'2亿行'（统计信息过期）；3）Spark根据'错误的统计信息'选择了Broadcast Join。

解决方案：我们运行了'ANALYZE TABLE orders COMPUTE STATISTICS'更新统计信息，Spark重新选择了SortMergeJoin，查询延迟从3分钟降到20秒。之后我们配置了'定时任务'每天凌晨更新统计信息。

# 定时任务：每天凌晨更新统计信息
# crontab -e
0 2 * * * spark-sql -e "ANALYZE TABLE orders COMPUTE STATISTICS;"
0 2 * * * spark-sql -e "ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS user_id, product_id;"

总结

数据湖查询慢通常是'配置不当'导致的，不是数据湖本身的问题。核心优化是：启用谓词下推（减少扫描数据量）、启用向量化执行（提升CPU效率）、定期更新统计信息（CBO选择最优执行计划）、合理的分区设计（减少小文件）。验证方法是用'df.explain(True)'查看执行计划，确认'谓词下推和向量化执行'是否生效。面试时能说清楚'为什么会慢、怎么配置参数、实战调优经验'，就很稳。

下一篇进入新的主题：OLAP引擎对比——Doris、StarRocks、ClickHouse、Hologres怎么选？

你们生产环境遇到过数据湖查询慢的问题吗？怎么解决的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长