4、数据交换实战：基于DataX实现MySQL到Hive的全量/增量同步

数据交换这件事，说白了就是让数据在不同系统之间「搬家」。我做了这么多年数据架构，见过太多团队在搬家过程中翻车——要么丢数据，要么跑不动，要么同步完才发现对不上。

今天咱们聊的，是数据交换里最经典的一个场景：从MySQL同步到Hive。MySQL存业务数据，Hive做离线分析，这几乎是每家公司都会遇到的组合。工具我选DataX，阿里开源的，稳定、灵活、社区活跃。我个人习惯用它来处理这类结构化数据的批量同步。

核心要点：全量同步解决「历史数据一次性搬迁」，增量同步解决「新数据持续追加」。两者结合，才能构建一个完整的数据同步链路。

4.1 场景分析：为什么是MySQL到Hive？

先说说这个场景的典型特征。MySQL是OLTP系统，存的是订单、用户、商品这类业务数据。Hive是OLAP系统，跑的是离线报表、数据挖掘。两者的数据模型、存储引擎、查询方式完全不同。

我在项目中遇到过这样一个情况：业务方每天要跑一份全量报表，但MySQL里几千万条数据，直接查会把数据库拖死。解决方案就是——每天凌晨把MySQL数据同步到Hive，然后在Hive上跑分析。这样既不影响线上业务，又能拿到完整数据。

对比维度 MySQL Hive
存储引擎 InnoDB/MyISAM HDFS + MapReduce/Tez
数据量级 GB ~ TB TB ~ PB
查询延迟 毫秒级 分钟级
适用场景 在线事务处理 离线批量分析
同步方式 源端 目标端

你想想看，这两个系统天然就不适合直连查询。所以数据交换就成了必经之路。DataX在这条路上，扮演的就是「搬运工」的角色。

4.2 DataX核心原理速览

DataX的架构其实不复杂。它把数据同步抽象成三个角色：Reader、Channel、Writer。Reader从源端读数据，Writer往目标端写数据，Channel在中间做缓冲和传输。

嗯，这里要注意：DataX不是简单的「读一条写一条」。它内部有并发控制、断点续传、脏数据管理这些机制。我刚开始用的时候，以为它就是封装了个JDBC连接池，后来看了源码才发现，它的Channel设计挺讲究的——支持限流、支持内存控制、支持数据转换。

小提示：DataX的Channel数量直接影响同步速度。我个人习惯先设成CPU核数的2倍，然后根据实际跑出来的吞吐量再调。别一上来就设100个，容易把源库打挂。

下面这张图，是我画的一个简化版DataX同步流程，你可以直观感受一下：

DataX 同步流程框架 MySQL Reader Channel 传输 Hive Writer JDBC 连接池 限流 & 断点续传 HDFS 写入 Reader 负责读取，Writer 负责写入，Channel 负责缓冲和传输 三者通过 JSON 配置文件描述，DataX 自动调度执行

4.3 全量同步实战

全量同步，就是一次性把MySQL里的所有数据搬到Hive。听起来简单，但坑不少。我见过有人直接SELECT * FROM table然后往HDFS写文件，结果源库内存直接爆了。

正确的做法是：分页读取 + 批量写入。DataX天然支持这个机制，你只需要配好JSON就行。

4.3.1 全量同步配置示例

下面这个配置，是我在项目里用过的简化版。它从MySQL的orders表读取数据，写入Hive的ods_orders表。

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "root",
            "password": "your_password",
            "connection": [
              {
                "jdbcUrl": ["jdbc:mysql://localhost:3306/biz_db"],
                "table": ["orders"]
              }
            ],
            "splitPk": "id",
            "where": "create_time >= '2024-01-01'"
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://namenode:8020",
            "fileType": "text",
            "path": "/user/hive/warehouse/ods_orders",
            "fileName": "orders",
            "writeMode": "overwrite",
            "column": [
              {"name": "id", "type": "bigint"},
              {"name": "order_no", "type": "string"},
              {"name": "amount", "type": "double"},
              {"name": "create_time", "type": "string"}
            ]
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 4
      }
    }
  }
}

注意：splitPk这个参数很关键。它告诉DataX用哪个字段做分片，实现并行读取。如果不设，DataX就只能单线程读，大表会慢得让你怀疑人生。我曾经踩过这个坑，一张2亿条的表跑了6个小时，加上splitPk后缩到40分钟。

4.3.2 全量同步的执行与验证

配置写好后，执行就一行命令：

python datax.py job/orders_full.json

跑完之后，我建议你做三件事验证：

  1. 行数对比：MySQL里SELECT COUNT(*)，Hive里也跑一下，看是否一致。
  2. 抽样对比：随机抽几条记录，逐字段比对，特别是时间、金额这种容易出问题的字段。
  3. 边界检查：看看NULL值、空字符串、特殊字符是否被正确处理。

我记得有一次，全量同步跑完行数对上了，但业务方反馈数据不对。查了半天，发现是MySQL里有个字段存的是JSON类型，Hive那边当字符串存了，但引号被转义了。从那以后，我每次都会加一个「字段类型映射检查」的步骤。

4.4 增量同步实战

全量同步搞定后，接下来就是增量同步。增量同步的核心是：只同步新增或变更的数据，而不是每次都全量搬一遍。

增量同步的方案，我总结下来主要有三种：

方案 原理 优点 缺点
时间戳增量 根据update_time字段筛选 实现简单，对源库影响小 依赖业务字段，删除数据无法感知
Binlog监听 解析MySQL的binlog日志 实时性高，能捕获所有变更 需要开启binlog，运维成本高
全量对比 全量拉取后在Hive侧做diff 数据最准确 资源消耗大，不适合大表

我个人最常用的是时间戳增量。原因很简单——大部分业务表都有create_time和update_time字段，而且DataX天然支持where条件过滤。你只需要记录上次同步的时间点，然后每次只拉取这个时间点之后的数据。

4.4.1 增量同步配置示例

增量同步的配置和全量很像，区别在于：

  • writeMode从overwrite改成append
  • where条件里加上时间过滤
{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "root",
            "password": "your_password",
            "connection": [
              {
                "jdbcUrl": ["jdbc:mysql://localhost:3306/biz_db"],
                "table": ["orders"]
              }
            ],
            "splitPk": "id",
            "where": "update_time >= '2024-06-01 00:00:00'"
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://namenode:8020",
            "fileType": "text",
            "path": "/user/hive/warehouse/ods_orders",
            "fileName": "orders_incr",
            "writeMode": "append",
            "column": [
              {"name": "id", "type": "bigint"},
              {"name": "order_no", "type": "string"},
              {"name": "amount", "type": "double"},
              {"name": "create_time", "type": "string"},
              {"name": "update_time", "type": "string"}
            ]
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 2
      }
    }
  }
}

经验之谈：增量同步的Channel数我一般设得比全量小。因为增量数据量小，并发太高反而浪费资源。另外，时间戳字段一定要建索引，否则每次同步都会全表扫描，源库扛不住。

4.4.2 增量同步的调度与状态管理

增量同步不是跑一次就完事的，它需要持续运行。我通常的做法是：

  1. 用一个状态表记录每次同步的max(update_time)
  2. 每次调度时，从状态表里读出上次同步的时间点
  3. 把这个时间点作为where条件的起始值
  4. 同步完成后，更新状态表

状态表的结构很简单：

CREATE TABLE sync_status (
  table_name VARCHAR(100) PRIMARY KEY,
  last_sync_time DATETIME,
  sync_type VARCHAR(10)  -- 'full' 或 'incr'
);

调度脚本里，我会先查这个表，然后动态生成DataX的JSON配置。嗯，这里要注意：时间边界要处理好。比如上次同步到2024-06-01 00:00:00，这次同步的where条件应该是update_time >= '2024-06-01 00:00:00'，而不是>。否则刚好在边界上的数据会漏掉。

4.5 全量+增量组合策略

在实际项目中，全量和增量不是二选一，而是组合使用。我常用的策略是：

  • 首次运行：全量同步，把历史数据一次性搬过来
  • 日常运行：增量同步，每天拉取新增/变更数据
  • 定期全量：比如每周或每月跑一次全量，用来修正增量同步可能产生的数据偏差

为什么会需要定期全量？因为增量同步有个天然缺陷——删除操作无法感知。如果MySQL里删了一条记录，增量同步不会主动去Hive里删对应的数据。时间一长，两边数据就对不上了。定期全量相当于一次「数据校准」。

最佳实践：我建议每天增量 + 每周全量。增量保证时效性，全量保证准确性。如果数据量特别大，全量可以放在周末凌晨跑，避开业务高峰期。

4.6 避坑指南

最后，分享几个我在实战中踩过的坑，希望能帮你少走弯路。

  • 字符集问题：MySQL用UTF-8，Hive默认也是UTF-8，但中间经过DataX传输时，如果配置不对，中文会变成乱码。解决方案：在JDBC URL里加上characterEncoding=utf-8。
  • 大字段处理：MySQL里的TEXT、BLOB字段，同步到Hive后可能变成空值。这是因为DataX默认的byteBuffer大小不够。调大byteBufferSize参数即可。
  • 时间精度丢失：MySQL的datetime(3)存的是毫秒级，Hive的string类型存成字符串后，毫秒部分可能被截断。我建议统一用yyyy-MM-dd HH:mm:ss.SSS格式。
  • 内存溢出：如果一张表有几百个字段，DataX默认的堆内存可能不够。启动时加上-Xmx4g参数，给足内存。

我曾经在一个项目里，因为没注意字符集问题，导致同步到Hive的中文数据全是???。业务方第二天跑报表，发现所有中文用户名都变成了问号，那叫一个尴尬。从那以后，字符集检查成了我每次同步前的必做项。

← 上一章 📖 返回目录 下一章 →