Sqoop 大表增量导入策略

一、为什么需要增量导入

全量导入大表（亿级数据）每次从头来代价极大，而且业务数据是持续增长的：

▼
text
复制代码
全量导入：每次导全表 → 1亿行 → 耗时数小时 → 每天做一次不现实

增量导入：只导新增/变更的数据 → 10万行 → 耗时几分钟 → 每天甚至每小时做一次

Sqoop 通过 --incremental 参数实现增量导入，提供两种策略。

---

二、两种增量导入策略

策略一：append 模式（基于自增列）

适用场景：表有自增主键，且只有新增操作，没有更新操作。

▼
text
复制代码
源表 users：
┌──────┬───────┬─────┐
│ id   │ name  │ age │
├──────┼───────┼─────┤
│ 1    │ alice │ 25  │  ← 上次已导入
│ 2    │ bob   │ 30  │  ← 上次已导入
│ ...  │ ...   │ ... │
│ 10000│ carol │ 28  │  ← 上次已导入（last-value=10000）
│ 10001│ dave  │ 22  │  ← 新增！本次需要导入
│ 10002│ eve   │ 35  │  ← 新增！本次需要导入
└──────┴───────┴─────┘

Sqoop 实际执行的 SQL：

▼
sql
复制代码
SELECT id, name, age FROM users
WHERE id > 10000           -- 只取比上次最大的 id 更大的行

命令示例：

▼
bash
复制代码
sqoop import \
  --connect jdbc:mysql://localhost:3306/mydb \
  --username root \
  --password 123456 \
  --table users \
  --incremental append \
  --check-column id \
  --last-value 10000 \
  --target-dir /data/users

参数	作用
--incremental append	增量模式：追加，只导入 check-column 大于 last-value 的行
--check-column id	用哪一列判断增量（通常是自增主键）
--last-value 10000	上次导入时该列的最大值，本次只导入比它大的

执行后：Sqoop 会输出本次导入的最大 id（如 10002），作为下次的 --last-value。

---

策略二：lastmodified 模式（基于时间戳列）

适用场景：表既有新增也有更新操作，有一个记录最后修改时间的列。

▼
text
复制代码
源表 orders：
┌──────┬─────────┬────────────────────┐
│ id   │ status  │ update_time        │
├──────┼─────────┼────────────────────┤
│ 1    │ paid    │ 2024-01-01 10:00   │  ← 上次已导入
│ 2    │ shipped │ 2024-01-01 11:00   │  ← 上次已导入（last-value=2024-01-01 12:00:00）
│ 1    │ done    │ 2024-01-02 09:00   │  ← 更新了！status 从 paid 变成 done
│ 3    │ new     │ 2024-01-02 10:00   │  ← 新增！
└──────┴─────────┴────────────────────┘

Sqoop 实际执行的 SQL：

▼
sql
复制代码
SELECT id, status, update_time FROM orders
WHERE update_time >= '2024-01-01 12:00:00'   -- 取上次时间点之后的所有变更

命令示例：

▼
bash
复制代码
sqoop import \
  --connect jdbc:mysql://localhost:3306/mydb \
  --username root \
  --password 123456 \
  --table orders \
  --incremental lastmodified \
  --check-column update_time \
  --last-value "2024-01-01 12:00:00" \
  --merge-key id \
  --target-dir /data/orders

参数	作用
--incremental lastmodified	增量模式：按时间戳，导入指定时间点之后的所有变更（含更新）
--check-column update_time	用哪一列判断增量（通常是 timestamp 类型）
--last-value	上次导入的时间点
--merge-key id	关键参数：用哪一列做合并键，新旧数据按此键去重合并

--merge-key 的作用：lastmodified 模式会导入"变更后的新行"，但 HDFS 中已有"旧行"。Sqoop 启动一个额外的 MapReduce Reduce 阶段，按 --merge-key 合并新旧数据，用新行覆盖旧行。

---

三、两种策略对比

	append 模式	lastmodified 模式
判断依据	自增列（如 id）	时间戳列（如 update_time）
能捕获更新	❌ 不能，只能捕获新增	✅ 能，新增和更新都能捕获
目标存储	只能追加到 HDFS	需要合并覆盖（依赖 --merge-key）
额外 MR 阶段	无，纯 Map 导入	有，Reduce 阶段做合并
性能	快，只做追加	较慢，需要读取旧数据 + 合并
适用表类型	日志表、只增不改的流水表	业务表，有增有改的状态表

---

四、last-value 的管理问题

增量导入的核心难点：如何记住上次的 last-value？

方式一：手动管理（简单但易错）

每次导入后，从 Sqoop 输出日志中找到新的 last-value，手动填到下次命令中。

▼
text
复制代码
缺点：人工操作，容易遗漏或填错

方式二：saved job（推荐）

Sqoop 内置作业保存功能，自动维护 last-value：

▼
bash
复制代码
# 1. 保存作业（只需执行一次）
sqoop job --create my_incremental_import \
  -- import \
  --connect jdbc:mysql://localhost:3306/mydb \
  --username root \
  --password 123456 \
  --table users \
  --incremental append \
  --check-column id \
  --last-value 0 \
  --target-dir /data/users

# 2. 每次执行（Sqoop 自动读取并更新 last-value）
sqoop job --exec my_incremental_import

执行流程：

▼
text
复制代码
sqoop job --exec my_incremental_import
      │
      ▼
  读取保存的 last-value（如 10000）
      │
      ▼
  执行导入：WHERE id > 10000
      │
      ▼
  导入完成，获取新的最大 id（如 10002）
      │
      ▼
  自动更新 saved job 中的 last-value = 10002
      │
      ▼
  下次执行时自动从 10002 开始

注意：--password 明文写在 saved job 中有安全风险。生产环境通常配合 --password-file 使用，密码存在 HDFS 上的受保护文件中。

---

五、增量导入的常见问题与解决方案

问题 1：边界数据丢失

▼
text
复制代码
上次 last-value = 10000
本次导入时，正好有 id=10000 的行在导入过程中被插入

Sqoop 执行：WHERE id > 10000  →  id=10000 的行被跳过！

解决方案：lastmodified 模式用 >= 而非 >，允许少量重叠。append 模式可在业务层保证"导入期间不写入"或使用 >= 并在目标端去重。

问题 2：时间精度问题

▼
text
复制代码
update_time 精度到秒，同一秒内有多条变更

last-value = 2024-01-01 12:00:00
WHERE update_time >= '2024-01-01 12:00:00'
  → 同一秒内已导入的数据会被重复导入

解决方案：使用毫秒级时间戳列，或在目标端做去重处理。

问题 3：数据延迟

▼
text
复制代码
业务系统写入数据库 → 有几秒延迟 → 增量导入时还没写入 → 漏掉

解决方案：增量导入时间窗口留出余量，比如 last-value 回退几分钟，宁可重复不可遗漏。

问题 4：lastmodified 模式的合并性能

▼
text
复制代码
合并需要读取 HDFS 中的旧数据 + 新导入的数据 → Reduce 阶段合并
  → 大表合并非常慢
  → 每次增量都要全量读取旧数据

解决方案：

方案	说明
导入到 Hive 分区表	按日期分区，每次增量只合并当天分区，不用读全表
改用 HBase	HBase 天然支持按 RowKey 覆盖更新，不需要合并
改用 Kudu	Kudu 支持 upsert 操作，适合增量更新场景

---

六、生产环境最佳实践

▼
text
复制代码
增量导入的完整生产方案：

  1. 选择策略
     ├── 只增不改的表 → append + 自增主键
     └── 有增有改的表 → lastmodified + 时间戳列

  2. 管理 last-value
     └── 用 sqoop job --create 保存，自动维护

  3. 调度执行
     └── 用 Oozie / Airflow / Crontab 定时执行 sqoop job --exec

  4. 防止遗漏
     ├── last-value 回退几分钟（宁可重复不可遗漏）
     └── 目标端做去重（Hive 用 ROW_NUMBER / HBase 用 RowKey 覆盖）

  5. 监控告警
     ├── 检查每次导入的行数，异常波动告警
     └── 检查 last-value 是否正常推进

---

七、总结

▼
text
复制代码
Sqoop 增量导入的本质：

  append 模式：      WHERE id > last_value
                    → 简单高效，但只能捕获新增

  lastmodified 模式：WHERE update_time >= last_value + merge 合并
                    → 能捕获更新，但需要额外合并开销

  核心机制：用 --check-column 和 --last-value 划定增量边界
  核心难点：last-value 的持久化管理 → 用 saved job 自动维护
  核心风险：边界遗漏 → 回退时间窗口 + 目标端去重兜底

Sqoop 的增量导入是"基于查询条件的增量拉取"——它不监听数据库变更（不像 CDC/Binlog 方案），而是通过"记住上次的位置 + 查询比它新的数据"来实现增量。这种方式简单通用，但实时性不如 Flink CDC / Canal 等 Binlog 订阅方案。