Sqoop 1.4.7 数据迁移：MySQL与HDFS双向同步实战

一、技术原理

Sqoop 通过 MapReduce 实现关系型数据库与 HDFS 的数据传输：

• 导入：MySQL $\to$ HDFS（sqoop import）
• 导出：HDFS $\to$ MySQL（sqoop export） 双向同步需结合增量策略： $$ \Delta D_{\text{同步}} = \text{CDC}(\Delta D_{\text{MySQL}}) \cup \text{Timestamp}(\Delta D_{\text{HDFS}}) $$ 其中 CDC 为变更数据捕获，$\Delta D$ 表示增量数据。

---

二、环境准备

1. 组件版本：
   • Sqoop 1.4.7
   • Hadoop 3.x
   • MySQL 8.0
2. MySQL 配置：

   CREATE USER 'sqoop'@'%' IDENTIFIED BY 'password';
   GRANT ALL PRIVILEGES ON source_db.* TO 'sqoop'@'%';
   

3. Sqoop 驱动：

   cp mysql-connector-java-8.0.28.jar $SQOOP_HOME/lib
   

---

三、MySQL $\to$ HDFS 同步（增量导入）

sqoop import \
--connect jdbc:mysql://mysql-host:3306/source_db \
--username sqoop \
--password password \
--table orders \
--target-dir /user/hadoop/orders \
--incremental append \  # 增量模式
--check-column order_id \  # 增量依据列
--last-value 1000  # 上次同步最大值

关键参数：

• --incremental：支持 append（自增ID）或 lastmodified（时间戳）
• --merge-key：合并数据时的主键（需配合 --incremental lastmodified）

---

四、HDFS $\to$ MySQL 同步（全量/增量导出）

sqoop export \
--connect jdbc:mysql://mysql-host:3306/target_db \
--username sqoop \
--password password \
--export-dir /user/hadoop/orders_processed \
--table orders \
--update-key order_id \  # 根据主键更新
--update-mode allowinsert  # 允许插入新记录

增量策略：

1. HDFS 数据分区按时间戳存储：/data/yyyyMMdd
2. 通过脚本每日导出新增分区：

   export_date=$(date +%Y%m%d)
   sqoop export --export-dir /data/$export_date ...
   

---

五、双向同步方案

graph LR
  A[MySQL] -- 定时增量导入 --> B[HDFS]
  B -- MapReduce处理 --> C[处理结果]
  C -- 定时增量导出 --> A

调度实现：

1. 使用 Oozie 或 Airflow 编排任务
2. 同步周期配置示例（每日同步）：

   <workflow-app name="sqoop-sync">
     <action name="mysql-to-hdfs">
       <sqoop xmlns="uri:oozie:sqoop-action:0.4">
         <command>import --connect jdbc:mysql://... </command>
       </sqoop>
     </action>
     <action name="hdfs-to-mysql" depends="mysql-to-hdfs">
       <sqoop>
         <command>export --export-dir /user/... </command>
       </sqoop>
     </action>
   </workflow-app>
   

---

六、常见问题解决

1. 主键冲突：
   • 导出时添加 --update-key
   • MySQL 表需设置主键
2. 数据类型映射：
   • 使用 --map-column-java 指定类型

   --map-column-java price=Double,create_time=String
   

3. 性能优化：
   • 调整 Map 并行度：-m 8
   • 启用压缩：--compress --compression-codec org.apache.hadoop.io.compress.SnappyCodec

注意：双向同步需确保数据幂等性，建议在业务低峰期执行，并在 MySQL 启用 binlog 用于异常恢复。