Sqoop数据迁移 - 关系型数据库与Hadoop之间的数据搬运工

数据仓库建设中，一个绕不开的任务是：把业务系统（MySQL、Oracle、PostgreSQL）中的数据搬运到 Hadoop 平台。Sqoop 就是专为这个场景设计的工具，负责关系型数据库（RDBMS）与 Hadoop 生态（HDFS、Hive、HBase）之间的数据双向迁移。

Sqoop 1 vs Sqoop 2

Sqoop 存在两个版本，生产环境绝大多数使用 Sqoop 1，本文聚焦 Sqoop 1：

核心架构：MapReduce 并行导入

Sqoop 的聪明之处在于把数据迁移任务转化为 MapReduce 作业，实现并行导入。

%%{init: {"theme":"base","themeVariables":{"fontFamily":"PingFang SC, -apple-system, Source Han Serif SC, serif","fontSize":"16px","background":"#FDFBF5","primaryColor":"#FDFBF5","primaryTextColor":"#2D2A26","primaryBorderColor":"#6E6A60","secondaryColor":"#E8EFE8","secondaryBorderColor":"#8FA890","secondaryTextColor":"#2D2A26","tertiaryColor":"#FDFBF5","tertiaryBorderColor":"#9A9488","tertiaryTextColor":"#6E6A60","lineColor":"#6E6A60","textColor":"#2D2A26","mainBkg":"#FDFBF5","nodeBorder":"#2D5A3D","clusterBkg":"#F3EEDF","clusterBorder":"#9A9488","edgeLabelBackground":"#FDFBF5"},"flowchart":{"htmlLabels":true,"curve":"basis","padding":20,"nodeSpacing":60,"rankSpacing":70,"useMaxWidth":false},"themeCSS":".node rect, .node polygon, .node circle { stroke-width: 1.5px; } .edgePath .path { stroke-width: 1.5px; } .cluster-label .nodeLabel, .cluster-label foreignObject { color: #2D2A26 !important; } .edgeLabel, .edgeLabel foreignObject { color: #2D2A26 !important; background-color: #FDFBF5; } .edgeLabel rect { fill: #FDFBF5; }"}}%%
flowchart LR
    classDef default fill:#FDFBF5,stroke:#6E6A60,color:#2D2A26,stroke-width:1.5px
    classDef startend fill:#2D2A26,stroke:#2D2A26,color:#FDFBF5,font-weight:bold
    classDef decision fill:#F6EAD8,stroke:#C08A1A,color:#2D2A26,font-weight:bold
    classDef hub fill:#3B5A80,stroke:#3B5A80,color:#FDFBF5,font-weight:bold
    classDef io fill:#E5EAF0,stroke:#3B5A80,color:#2D2A26
    classDef accent fill:#A6432C,stroke:#A6432C,color:#FDFBF5,font-weight:bold
    classDef storage fill:#F3EEDF,stroke:#9A9488,color:#2D2A26
    subgraph "数据源"
        RDBMS[关系型数据库\nMySQL / Oracle / PostgreSQL]
    end

    subgraph "Sqoop 客户端"
        CLI[sqoop import 命令]
        META[查询元数据\n表结构、主键、行数]
        GEN[生成 Java 代码\nMapper 类]
    end

    subgraph "MapReduce 并行导入"
        M1[Mapper 1\n处理 ID 1-250000]
        M2[Mapper 2\n处理 ID 250001-500000]
        M3[Mapper 3\n处理 ID 500001-750000]
        M4[Mapper 4\n处理 ID 750001-1000000]
    end

    subgraph "目标存储"
        HDFS[HDFS 目录\nor Hive 表]
    end

    CLI --> META
    META --> RDBMS
    META --> GEN
    GEN --> M1
    GEN --> M2
    GEN --> M3
    GEN --> M4
    M1 -->|JDBC 分片查询| RDBMS
    M2 -->|JDBC 分片查询| RDBMS
    M3 -->|JDBC 分片查询| RDBMS
    M4 -->|JDBC 分片查询| RDBMS
    M1 --> HDFS
    M2 --> HDFS
    M3 --> HDFS
    M4 --> HDFS
    class RDBMS,CLI,META,GEN,M1,M2,M3,M4,HDFS default

为什么用 MapReduce 并行导入：

• 单线程 JDBC 导出大表可能受吞吐限制（原稿“每秒几千行”为未核实的教学示例，不是通用性能基准）
• 将表按主键范围分成多个区间，每个 Mapper 处理一个区间，并行 JDBC 查询
• 原稿的“4 个 Mapper 理论上速度是单线程的 4 倍”仅作为未核实的理想线性加速示例；实际收益取决于分片均衡、数据库负载、网络和作业开销，不能保证四倍加速
• 导入结果作为多个文件写入 HDFS，自然支持分布式存储

安装配置

JDBC 驱动安装

Sqoop 本身不包含数据库驱动，需要手动将对应数据库的 JDBC 驱动 JAR 放入 Sqoop 的 lib 目录：

# MySQL 驱动安装
cp mysql-connector-java-8.0.30.jar $SQOOP_HOME/lib/

# Oracle 驱动安装（需从 Oracle 官网下载，有许可证限制）
cp ojdbc8.jar $SQOOP_HOME/lib/

# PostgreSQL 驱动安装
cp postgresql-42.5.0.jar $SQOOP_HOME/lib/

sqoop-env.sh 配置

# $SQOOP_HOME/conf/sqoop-env.sh

# 指向 Hadoop 安装目录（必须）
export HADOOP_COMMON_HOME=/opt/hadoop

# 指向 Hive 安装目录（需要导入 Hive 表时配置）
export HIVE_HOME=/opt/hive

# 指向 HBase 安装目录（需要导入 HBase 时配置）
export HBASE_HOME=/opt/hbase

# 指向 ZooKeeper 安装目录
export ZOOCFGDIR=/opt/zookeeper/conf

连接测试

# 列出 MySQL 数据库中的所有 Schema
sqoop list-databases \
  --connect jdbc:mysql://mysql-host:3306/ \
  --username sqoop_user \
  --password sqoop_password

# 列出某个数据库下的所有表
sqoop list-tables \
  --connect jdbc:mysql://mysql-host:3306/business_db \
  --username sqoop_user \
  --password sqoop_password

echo -n "your_password" | hdfs dfs -put - /user/sqoop/.sqoop_pwd
hdfs dfs -chmod 400 /user/sqoop/.sqoop_pwd
# 之后命令中使用
--password-file hdfs:///user/sqoop/.sqoop_pwd

核心命令详解

sqoop import：从 RDBMS 导入到 Hadoop

基础全量导入：

sqoop import \
  --connect jdbc:mysql://mysql-host:3306/business_db \
  --username sqoop_user \
  --password sqoop_password \
  --table orders \
  --target-dir /data/raw/orders \
  --num-mappers 4 \
  --split-by order_id \
  --fields-terminated-by '\t' \
  --delete-target-dir

--split-by 分片列选择原则：

--split-by 指定用于分割数据范围的列，Sqoop 会查询该列的 MIN 和 MAX 值，然后均匀划分成 num-mappers 个区间。选择原则：

sqoop import --incremental：增量导入

增量导入解决了"只同步新增/更新数据"的需求，有两种模式：

append 模式（追加新增行，基于自增 ID 或时间戳）：

sqoop import \
  --connect jdbc:mysql://mysql-host:3306/business_db \
  --username sqoop_user \
  --password sqoop_password \
  --table orders \
  --target-dir /data/raw/orders \
  --incremental append \
  --check-column order_id \
  --last-value 1000000 \
  --num-mappers 4

lastmodified 模式（同步更新行，基于时间戳列）：

sqoop import \
  --connect jdbc:mysql://mysql-host:3306/business_db \
  --username sqoop_user \
  --password sqoop_password \
  --table orders \
  --target-dir /data/raw/orders \
  --incremental lastmodified \
  --check-column updated_at \
  --last-value "2024-01-01 00:00:00" \
  --merge-key order_id \
  --num-mappers 4

两种增量模式对比：

sqoop export：从 Hadoop 导出到 RDBMS

# 基础导出：HDFS 文件导出到 MySQL 表
sqoop export \
  --connect jdbc:mysql://mysql-host:3306/report_db \
  --username sqoop_user \
  --password sqoop_password \
  --table ads_user_summary \
  --export-dir /data/output/ads_user_summary \
  --input-fields-terminated-by '\t' \
  --num-mappers 4

# Upsert 模式：有则更新，无则插入
sqoop export \
  --connect jdbc:mysql://mysql-host:3306/report_db \
  --username sqoop_user \
  --password sqoop_password \
  --table ads_user_summary \
  --export-dir /data/output/ads_user_summary \
  --update-key user_id \
  --update-mode allowinsert \
  --input-fields-terminated-by '\t' \
  --num-mappers 4

sqoop job：保存和复用增量作业

对于需要定期执行的增量导入，sqoop job 可以保存作业配置，并自动记录上次的 last-value，下次执行时从上次结束的位置继续：

# 创建增量导入作业
sqoop job --create import_orders_job \
  -- import \
  --connect jdbc:mysql://mysql-host:3306/business_db \
  --username sqoop_user \
  --password sqoop_password \
  --table orders \
  --target-dir /data/raw/orders \
  --incremental append \
  --check-column order_id \
  --last-value 0 \
  --num-mappers 4

# 列出所有已保存的作业
sqoop job --list

# 执行作业（自动更新 last-value）
sqoop job --exec import_orders_job

# 查看作业详情
sqoop job --show import_orders_job

# 删除作业
sqoop job --delete import_orders_job

常用参数速查表

生产实践

按日期分区导入 Hive

实际项目中，每天同步昨天的数据到 Hive 分区表是最常见的场景，通常通过 Shell 脚本驱动：

#!/bin/bash
# daily_import_orders.sh - 每日增量同步订单表到 Hive 分区

YESTERDAY=$(date -d "yesterday" +%Y-%m-%d)

sqoop import \
  --connect "jdbc:mysql://mysql-host:3306/business_db?useSSL=false&characterEncoding=UTF-8" \
  --username sqoop_user \
  --password-file hdfs:///user/sqoop/.sqoop_pwd \
  --query "SELECT order_id, user_id, amount, status, created_time
           FROM orders
           WHERE DATE(created_time) = '${YESTERDAY}'
           AND \$CONDITIONS" \
  --split-by order_id \
  --num-mappers 4 \
  --hive-import \
  --hive-table ods.ods_orders \
  --target-dir "/data/staging/ods_orders/dt=${YESTERDAY}" \
  --hive-partition-key dt \
  --hive-partition-value "${YESTERDAY}" \
  --hive-overwrite \
  --null-string '\N' \
  --null-non-string '\N' \
  --fields-terminated-by '\t' \
  --compress \
  --compression-codec snappy

if [ $? -eq 0 ]; then
    echo "[$(date)] 订单数据同步成功: ${YESTERDAY}"
else
    echo "[$(date)] 订单数据同步失败: ${YESTERDAY}" >&2
    exit 1
fi

性能调优

避坑清单

Sqoop vs DataX 选型对比

掌握检查

• [ ] 能解释 Sqoop 为什么使用 MapReduce 并行导入，--split-by 和 --num-mappers 参数如何影响并行度
• [ ] 理解 incremental append 和 lastmodified 两种增量模式的区别，知道各自的适用场景
• [ ] 能写出将 MySQL 表按日期分区导入 Hive 的 sqoop import 命令，包含正确的分区参数
• [ ] 知道 NULL 值处理、特殊字符编码等常见坑及解决方法

实践练习

基础题：写一条 sqoop import 命令，将 MySQL 数据库 sales_db 中的 orders 表全量导入到 HDFS 的 /data/raw/orders/dt=2024-01-15 目录。要求：4 个 Mapper，按 order_id 分片，字段用 \t 分隔，NULL 值替换为 \N。

进阶题：业务方要求每天凌晨 2 点将昨天新增的订单（created_time 字段）同步到 Hive 的 ods_orders 表对应日期分区。设计完整的增量同步方案：选择使用哪种增量模式？关键参数如何设置？如何保证幂等性（重跑时不产生重复数据）？用 Shell 脚本框架描述执行逻辑。

相关文档

• HDFS分布式文件系统 - Sqoop 导入数据的目标存储
• Hive数据仓库 - Sqoop 直接导入 Hive 表的集成方式
• MapReduce编程模型 - Sqoop 底层执行引擎的工作原理
• Hadoop集群规划 - 数据同步在整体平台架构中的位置
• ZooKeeper协调服务 - Sqoop 任务依赖的 Hadoop 集群协调基础