How to Sync 50 Million Rows Efficiently with Alibaba’s DataX

A project with 50 million rows faced inaccurate reporting data and cross‑database operations, making SQL‑based synchronization impractical. Attempts using mysqldump or file‑based storage proved too slow or ineffective.

DataX, the open‑source version of Alibaba Cloud DataWorks, is introduced as a robust solution for offline data synchronization across heterogeneous sources such as MySQL, Oracle, HDFS, Hive, ODPS, HBase, FTP, etc.

DataX Overview

DataX transforms complex mesh‑type sync chains into a star‑shaped data flow, acting as a middle‑layer that connects various data sources.

DataX 3.0 Framework Design

When a new data source is added, it can be seamlessly integrated with DataX.

DataX 3.0 Core Architecture

DataX executes a synchronization job (Job). After receiving a Job, it launches a process that handles data cleaning, task splitting, and TaskGroup management.

Job splits into multiple Tasks based on source split strategy for concurrent execution.

Scheduler assembles Tasks into TaskGroups according to the configured concurrency.

Each Task starts a Reader → Channel → Writer thread chain.

Job monitors TaskGroups and exits successfully when all complete (non‑zero exit code on failure).

DataX scheduling process:

Job splits the data into Tasks and calculates the number of TaskGroups based on user‑defined concurrency.

Task/Channel = TaskGroup; TaskGroups run the assigned Tasks.

Using DataX for Data Synchronization

Preparation

JDK 1.8+

Python 2 or 3 (CentOS 7 already includes Python 2.7)

Apache Maven 3.x (only needed for compiling DataX tar package)

Install JDK, extract it, and set JAVA_HOME and PATH in /etc/profile.

Download and extract DataX:

# wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
# tar zxf datax.tar.gz -C /usr/local/
# rm -rf /usr/local/datax/plugin/*/._*

Verify installation:

# cd /usr/local/datax/bin
# python datax.py ../job/job.json

Installing MySQL on Two Hosts

# yum -y install mariadb mariadb-server mariadb-libs mariadb-devel
# systemctl start mariadb
# mysql_secure_installation

Create database and table on both hosts:

CREATE DATABASE `course-study`;
CREATE TABLE `course-study`.t_member(ID int, Name varchar(20), Email varchar(30));

Grant privileges for DataX:

grant all privileges on *.* to root@'%' identified by '123123';
flush privileges;

Creating a Stored Procedure to Generate Test Data

DELIMITER $$
CREATE PROCEDURE test()
BEGIN
DECLARE A int default 1;
WHILE (A < 3000000) DO
INSERT INTO `course-study`.t_member VALUES(A, concat('LiSa',A), concat('LiSa',A,'@163.com'));
SET A = A + 1;
END WHILE;
END $$
DELIMITER ;

Call the procedure to populate data:

CALL test();

Full‑Load Synchronization with DataX

Generate a MySQL‑to‑MySQL job template (JSON) and customize it:

{
  "job": {
    "content": [{
      "reader": {
        "name": "mysqlreader",
        "parameter": {
          "column": ["*"],
          "connection": [{
            "jdbcUrl": ["jdbc:mysql://192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"],
            "table": ["t_member"]
          }],
          "username": "root",
          "password": "123123",
          "splitPk": "ID"
        }
      },
      "writer": {
        "name": "mysqlwriter",
        "parameter": {
          "column": ["*"],
          "connection": [{
            "jdbcUrl": ["jdbc:mysql://192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8"],
            "table": ["t_member"]
          }],
          "username": "root",
          "password": "123123",
          "preSql": ["truncate t_member"],
          "session": ["set session sql_mode='ANSI'"],
          "writeMode": "insert"
        }
      }
    }],
    "setting": {"speed": {"channel": "5"}}
  }
}

Run the job:

# python /usr/local/datax/bin/datax.py install.json

Sample output shows successful synchronization of 2,999,999 records at ~2.57 MB/s.

Incremental Synchronization

The only difference from full‑load is adding a where clause for conditional filtering and optionally removing preSql (e.g., truncate).

{
  "job": {
    "content": [{
      "reader": {
        "name": "mysqlreader",
        "parameter": {
          "column": ["*"],
          "splitPk": "ID",
          "where": "ID <= 1888",
          "connection": [{
            "jdbcUrl": ["jdbc:mysql://192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"],
            "table": ["t_member"]
          }],
          "username": "root",
          "password": "123123"
        }
      },
      "writer": {
        "name": "mysqlwriter",
        "parameter": {
          "column": ["*"],
          "connection": [{
            "jdbcUrl": ["jdbc:mysql://192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8"],
            "table": ["t_member"]
          }],
          "username": "root",
          "password": "123123",
          "preSql": ["truncate t_member"],
          "session": ["set session sql_mode='ANSI'"],
          "writeMode": "insert"
        }
      }
    }],
    "setting": {"speed": {"channel": "5"}}
  }
}

Running this job synchronizes only the rows matching the condition (e.g., 1,888 records) and confirms successful incremental sync.

In practice, full synchronization works for initial data loads, while incremental sync with where filters is essential for large‑scale or ongoing data pipelines to avoid interruptions.