4、增量查询基础语法：Spark SQL 与 DataFrame API 两种方式

语法这东西，说白了就是工具的使用说明书。你调优调得再好，如果连查询怎么写都不对，那一切都是白搭。我个人习惯把增量查询的语法分成两条路：一条是 Spark SQL，适合数据分析师和喜欢写 SQL 的同学；另一条是 DataFrame API，适合做工程的同学，灵活度更高。

今天我们就来把这两条路都走一遍。我会把我在项目中踩过的坑、用过的技巧，都揉进代码里。

4.1 核心前提：开启增量查询模式

不管用哪种方式，你都得先告诉 Hudi：“嘿，我要查增量数据了。” 这个开关就是 hoodie.datasource.query.type，默认值是 snapshot（快照读），我们要改成 incremental。

4.2 方式一：Spark SQL 增量查询

Spark SQL 的方式最直观。你只需要在查询时指定一个 hoodie 表源，然后带上几个关键参数就行。

4.2.1 基本语法

-- 查询 hudi 表的增量数据
SELECT *
FROM hudi_table
WHERE `_hoodie_commit_time` > '20240101000000'
  AND `_hoodie_commit_time` <= '20240102000000'

嗯，这里要注意：_hoodie_commit_time 是 Hudi 自动维护的元数据字段，记录的是这条数据是在哪个 commit 里写入的。你直接用它做过滤条件，Hudi 就能自动识别出你要做增量查询。

4.2.2 使用 Spark SQL 的完整示例

-- 1. 注册 Hudi 表（假设表路径为 /user/hudi/orders）
CREATE TEMPORARY VIEW orders_hudi
USING org.apache.hudi
OPTIONS (
  path = '/user/hudi/orders',
  hoodie.datasource.query.type = 'incremental',
  hoodie.datasource.query.incremental.format = 'latest_state'
);

-- 2. 查询增量数据
SELECT order_id, order_status, update_time
FROM orders_hudi
WHERE `_hoodie_commit_time` > '20240101000000'
  AND `_hoodie_commit_time` <= '20240102000000'
  AND order_status = 'shipped';

4.3 方式二：DataFrame API 增量查询

DataFrame API 更适合在代码里动态拼接查询条件。我一般在做 ETL 任务时用这种方式，因为可以灵活控制时间范围。

4.3.1 基本语法

import org.apache.hudi.DataSourceReadOptions
import org.apache.hudi.DataSourceWriteOptions
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("HudiIncrementalQuery")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

// 读取 Hudi 表的增量数据
val df = spark.read
  .format("org.apache.hudi")
  .option(DataSourceReadOptions.QUERY_TYPE.key(), 
          DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_VAL)
  .option(DataSourceReadOptions.BEGIN_INSTANTTIME.key(), "20240101000000")
  .option(DataSourceReadOptions.END_INSTANTTIME.key(), "20240102000000")
  .load("/user/hudi/orders")

df.show()

4.3.2 参数详解

4.4 两种方式的对比

你可能会问：到底用哪种好？我个人的经验是：

• Spark SQL：适合临时查询、数据分析、快速验证。写起来快，但参数配置不够灵活。
• DataFrame API：适合生产环境、定时任务、动态参数。可以结合变量、循环、条件判断，灵活度更高。

举个例子，我在做实时数仓的增量同步时，都是用 DataFrame API。因为每天要同步的时间范围是动态计算的，用 SQL 的话每次都要改语句，太麻烦了。

4.5 避坑指南

我曾经在项目中遇到过几个坑，今天一并分享给你：

• 时间格式必须严格：begin_instanttime 和 end_instanttime 必须是 yyyyMMddHHmmss 格式，少一位都不行。我试过传 20240101，结果 Hudi 直接报错。
• 不要用 _hoodie_commit_time 做字符串比较：虽然它看起来是字符串，但 Hudi 内部是按时间戳处理的。如果你用 > '2024-01-01' 这种格式，可能会出问题。
• 增量查询不支持跨表 JOIN：如果你需要 JOIN 其他表，建议先查出增量数据，再单独做 JOIN 操作。

4.6 知识体系图

下面这张图帮你理清增量查询的两种方式及其核心参数：

4.7 实战小贴士

最后，分享几个我在生产环境中用到的实用技巧：

• 用变量控制时间范围：在 DataFrame API 中，把 begin_instanttime 和 end_instanttime 定义成变量，方便调试和复用。
• 先查元数据再查数据：如果你不确定有哪些 commit 时间，可以先查 _hoodie_commit_time 的 distinct 值，确认范围后再做增量查询。
• 增量查询 + 分区过滤：如果表有分区，建议同时加上分区过滤条件，能大幅减少扫描的数据量。