跳到主要内容

DataX 离线数据同步指南

概述​

DataX 是阿里巴巴开源的异构数据源离线同步工具，致力于实现包括关系型数据库（MySQL、Oracle 等）、HDFS、Hive、ODPS、HBase、FTP 等各种异构数据源之间稳定高效的数据同步功能。

架构设计​

┌──────────────┐
│   Reader     │
│ (数据读取)   │
└──────────────┘
       ↓
┌──────────────┐
│    Frame     │
│ (核心框架)   │
└──────────────┘
       ↓
┌──────────────┐
│   Writer     │
│ (数据写入)   │
└──────────────┘

支持的数据库类型​

关系型数据库​

数据库ReaderWriter说明
MySQL✅✅最常用
Oracle✅✅需要 OCI 或 Instant Client
PostgreSQL✅✅完全支持
SQL Server✅✅Windows/Linux
DB2✅✅IBM 大型机
MariaDB✅✅MySQL 兼容

NoSQL 和大数据存储​

数据源ReaderWriter说明
HDFS✅✅Hadoop 文件系统
Hive✅✅数据仓库
HBase✅✅列式存储
MongoDB✅✅文档数据库
Elasticsearch✅✅搜索引擎

其他数据源​

  • FTP / SFTP
  • OSS（阿里云对象存储）
  • ODPS（阿里云 MaxCompute）
  • ClickHouse
  • Doris/StarRocks

快速开始​

安装 DataX​

# 下载 DataX
cd /tmp
wget https://github.com/alibaba/DataX/releases/download/v3.0.0/datax.tar.gz

# 解压
tar -xzf datax.tar.gz
sudo mv datax /opt/datax
sudo chown -R dolphinscheduler:dolphinscheduler /opt/datax

# 配置环境变量
export DATAX_HOME=/opt/datax
export PATH=$DATAX_HOME/bin:$PATH

验证安装​

python3 $DATAX_HOME/bin/datax.py --help

集成到 DolphinScheduler​

1. 配置环境变量​

在 dolphinscheduler_env.sh 中：

# DataX 配置
export DATAX_HOME=/opt/datax
export DATAX_PYTHON=/usr/bin/python3

2. 创建 Shell 任务​

在 DS Web UI 中创建 Shell 任务：

#!/bin/bash

# 加载环境变量
source /opt/dolphinscheduler/conf/dolphinscheduler_env.sh

# DataX 作业配置
JOB_CONFIG="/opt/datax/job/mysql_to_hive.json"

# 执行同步
$DATAX_HOME/bin/datax.py $JOB_CONFIG

# 检查结果
if [ $? -eq 0 ]; then
  echo "DataX 任务执行成功"
  exit 0
else
  echo "DataX 任务执行失败"
  exit 1
fi

典型应用场景​

MySQL → Hive 离线同步​

{
  "job": {
    "setting": {
      "speed": {
        "channel": 5
      },
      "errorLimit": {
        "record": 0,
        "percentage": 0.02
      }
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "root",
            "password": "password",
            "column": ["*"],
            "connection": [
              {
                "table": ["users"],
                "jdbcUrl": ["jdbc:mysql://192.168.192.111:3306/source_db"]
              }
            ],
            "where": "update_time >= '${last_sync_time}'"
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://namenode:9000",
            "fileType": "text",
            "path": "/user/hive/warehouse/users",
            "fileName": "users",
            "column": [
              {"name": "id", "type": "bigint"},
              {"name": "name", "type": "string"},
              {"name": "age", "type": "int"}
            ],
            "writeMode": "append",
            "fieldDelimiter": ","
          }
        }
      }
    ]
  }
}

Oracle → MySQL 迁移​

{
  "job": {
    "setting": {
      "speed": {
        "channel": 3
      }
    },
    "content": [
      {
        "reader": {
          "name": "oraclereader",
          "parameter": {
            "username": "scott",
            "password": "tiger",
            "column": ["*"],
            "connection": [
              {
                "table": ["EMP"],
                "jdbcUrl": "jdbc:oracle:thin:@192.168.192.100:1521:orcl"
              }
            ]
          }
        },
        "writer": {
          "name": "mysqlwriter",
          "parameter": {
            "username": "root",
            "password": "password",
            "writeMode": "insert",
            "column": ["id", "ename", "job", "mgr", "hiredate", "sal", "comm", "deptno"],
            "connection": [
              {
                "jdbcUrl": "jdbc:mysql://192.168.192.111:3306/target_db",
                "table": ["emp"]
              }
            ]
          }
        }
      }
    ]
  }
}

PostgreSQL → Doris 实时数仓​

{
  "job": {
    "setting": {
      "speed": {
        "channel": 4
      }
    },
    "content": [
      {
        "reader": {
          "name": "postgresqlreader",
          "parameter": {
            "username": "postgres",
            "password": "password",
            "column": ["id", "name", "amount", "create_time"],
            "connection": [
              {
                "jdbcUrl": "jdbc:postgresql://192.168.192.111:5432/source_db",
                "table": ["orders"]
              }
            ]
          }
        },
        "writer": {
          "name": "doriswriter",
          "parameter": {
            "username": "root",
            "password": "password",
            "column": ["id", "name", "amount", "create_time"],
            "preSql": ["TRUNCATE TABLE orders"],
            "connection": [
              {
                "jdbcUrl": "jdbc:mysql://192.168.192.140:9030/target_db",
                "table": ["orders"]
              }
            ]
          }
        }
      }
    ]
  }
}

性能优化​

1. 并发控制​

{
  "setting": {
    "speed": {
      "channel": 5,        // 并发通道数
      "byte": 104857600    // 每秒字节数限制（100MB/s）
    }
  }
}

2. 错误处理​

{
  "setting": {
    "errorLimit": {
      "record": 100,       // 允许错误的记录数
      "percentage": 0.01   // 错误率阈值（1%）
    }
  }
}

3. JVM 调优​

# 修改 datax.py 中的 JVM 参数
export JAVA_OPTS="-Xms4g -Xmx8g -XX:+HeapDumpOnOutOfMemoryError"

监控与告警​

日志位置​

/opt/datax/log/datax.log          # 主日志
/opt/datax/log/performance.log    # 性能日志
/opt/datax/log/stdout.log         # 标准输出

监控指标​

  • 同步速度（records/sec, bytes/sec）
  • 错误记录数
  • 任务执行时间
  • 资源使用率（CPU、内存）

最佳实践​

1. 增量同步​

使用时间戳字段进行增量同步：

# 获取上次同步时间
LAST_SYNC=$(psql -h pg_host -U postgres -d source_db -c \
  "SELECT COALESCE(MAX(update_time), '2000-01-01') FROM sync_log" -t)

# 替换配置文件中的变量
sed -i "s/\${last_sync_time}/$LAST_SYNC/g" job_config.json

# 执行同步
datax.py job_config.json

# 记录本次同步时间
psql -h pg_host -U postgres -d source_db -c \
  "INSERT INTO sync_log VALUES (NOW())"

2. 数据一致性校验​

-- 源表统计
SELECT COUNT(*), SUM(amount) FROM source_table;

-- 目标表统计
SELECT COUNT(*), SUM(amount) FROM target_table;

-- 对比结果

3. 批量作业管理​

#!/bin/bash

# 定义作业列表
JOBS=("job1.json" "job2.json" "job3.json")

# 并行执行
for job in "${JOBS[@]}"; do
  python3 $DATAX_HOME/bin/datax.py $job &
done

# 等待所有任务完成
wait

echo "所有作业执行完成"

常见问题​

Q1: DataX 和 SeaTunnel 选哪个？​

答： 根据场景选择：

  • DataX:

    • ✅ 离线同步（批量）
    • ✅ 成熟稳定（2015 年开源）
    • ✅ 阿里生态
    • ❌ 不支持实时 CDC

  • SeaTunnel:

    • ✅ 实时同步（CDC）
    • ✅ 高性能
    • ✅ 多表/复杂转换
    • ✅ 新项目首选

Q2: 如何处理大字段？​

{
  "reader": {
    "parameter": {
      "fetchSize": 1000,     // 分批读取
      "querySql": ["SELECT id, name FROM table WHERE condition"]
    }
  }
}

Q3: 如何保证数据一致性？​

  1. 使用事务（如果目标库支持）
  2. 先写入临时表，再切换
  3. 同步后进行数据校验
  4. 记录同步日志和位点

参考资源​