DataX est un outil open source de synchronisation de données développé par Alibaba. Il est très populaire sur GitHub avec plus de 14,8k étoiles. Sa page officielle se trouve à l'adresse : https://github.com/aliyuncs/DataX.

Qu'est-ce que DataX ?

DataX est la version open source du module d'intégration de données DataWorks d'Alibaba Cloud. Écrit en Java, il est largement utilisé au sein du groupe Alibaba comme outil de synchronisation de données hors ligne. DataX prend en charge la synchronisation efficace de données entre divers systèmes hétérogènes, notamment MySQL, Oracle, OceanBase, SQL Server, PostgreSQL, HDFS, Hive, ADS, HBase, TableStore (OTS), MaxCompute (ODPS), Hologres, DRDS, Databend, et bien d'autres.

Quels sont les cas d'usage ?

1. Synchronisation d'entrepôts de données : DataX permet de synchroniser des données d'un entrepôt (base relationnelle, système de stockage Big Data, etc.) vers un autre, pour des opérations de migration, sauvegarde ou réplication.
2. Migration de bases de données : Il facilite le transfert et la transformation de données entre différentes plateformes de bases de données.
3. Intégration et synchronisation de données : DataX peut agréger et synchroniser des données provenant de multiples sources (bases relationnelles, NoSQL, systèmes de fichiers, etc.) vers une destination unique.
4. Nettoyage et transformation de données : Il offre des capacités avancées de transformation : filtrage, mapping, conversion de formats, etc. Cela est crucial pour améliorer la qualité et la cohérence des données dans les entrepôts, lacs de données ou datamarts.
5. Sauvegarde et restauration de données : En configurant des tâches planifiées, DataX peut automatiser la sauvegarde des données depuis une source vers une cible, et permettre leur restauration si nécessaire.

Quelles sources de données DataX prend-il en charge ?

La liste complète est disponible sur le dépôt GitHub officiel.

Architecture

DataX est construit selon une architecture Framwork + plugin. La lecture et l'écriture des données sont abstraites sous forme de plugins Reader et Writer, intégrés au framework de synchronisation.

• Reader : Module de collecte. Il lit les données depuis la source et les transmet au Framework.
• Writer : Module d'écriture. Il reçoit les données du Framework et les écrit dans la destination.
• Framework : Assure la liaison entre le Reader et le Writer, gère le canal de transmission, le buffer, le contrôle de flux, la concurrence et la transformation des données.

La version open source de DataX fonctionne en mode mono-processus multi-thread pour exécuter les jobs de synchronisation.

1. Un job DataX (Job) est l'unité de synchronisation complète. Lorsqu'il reçoit un job, DataX lance un processus. Le module Job agit comme le nœud de gestion central : il gère le nettoyage des données, la division des sous-tâches (transformation d'un job unique en plusieurs Task) et la gestion des groupes de tâches (TaskGroup).
2. Le Job divise le travail en plusieurs Task (sous-tâches) selon la stratégie de partitionnement de la source, permettant une exécution parallèle. Chaque Task est l'unité atomique de synchronisation.
3. Ensuite, le module Scheduler du Job regroupe les Tasks en TaskGroup selon le niveau de concurrence configuré. Chaque TaskGroup exécute ses Tasks avec une concurrence fixe (par défaut 5).
4. Chaque Task est démarré par son TaskGroup. Une fois lancé, il instancie les threads Reader → Channel → Writer pour effectuer la synchronisation.
5. Le Job surveille la progression des TaskGroup. Si tous les TaskGroup terminent avec succès, le Job se termine normalement. Sinon, il se termine en erreur (code de sortie non nul).

Exemple de flux de planification

Prenons un exemple concret : un utilisateur soumet un job avec 20 canaux de concurrence, pour synchroniser 100 tables MySQL vers ODPS. La décision de DataX est la suivante :

• Le Job divise le travail en 100 Tasks (une par table).
• Avec 20 canaux, DataX calcule qu'il faut 4 TaskGroup (20 / 5).
• Les 4 TaskGroup se répartissent les 100 Tasks : chaque TaskGroup gère 5 tâches concurrentes, soit 25 tasks au total.

Comment utiliser DataX ?

Téléchargez DataX depuis la page de sortie officielle, puis décompressez-le dans un répertoire local.

Exemple de cas

Pour cet exemple, nous allons synchroniser la table user_info de MySQL vers la table ods_test_mysql_user_info_m (même base), avec un filtre sur le champ update_time.

-- Scripts de création des tables
DROP TABLE IF EXISTS ods_test_mysql_user_info_m;

CREATE TABLE `user_info` (
  `id` int NOT NULL COMMENT 'ID',
  `name` varchar(50) NOT NULL COMMENT 'Nom',
  `sex` tinyint NOT NULL COMMENT 'Genre (1: Homme, 2: Femme)',
  `phone` varchar(11) COMMENT 'Téléphone',
  `address` varchar(1000) COMMENT 'Adresse',
  `age` int COMMENT 'Âge',
  `create_time` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT 'Date de création',
  `update_time` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6) COMMENT 'Date de modification',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COMMENT='Table des informations utilisateur';

CREATE TABLE `ods_test_mysql_user_info_m` (
  `id` int NOT NULL COMMENT 'ID',
  `name` varchar(50) NOT NULL COMMENT 'Nom',
  `sex` tinyint NOT NULL COMMENT 'Genre',
  `phone` varchar(11) COMMENT 'Téléphone',
  `address` varchar(1000) COMMENT 'Adresse',
  `age` int COMMENT 'Âge',
  `create_time` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT 'Date de création',
  `update_time` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6) COMMENT 'Date de modification',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COMMENT='Table d'entrepôt des informations utilisateur';

Insérez ensuite des données d'exemple dans user_info.

Créer le fichier de configuration JSON du job

Dans le répertoire script de DataX, créez le fichier sync_user_info.json avec le contenu suivant. mysqlreader est le plugin de lecture, mysqlwriter celui d'écriture.

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": ["id","name","sex","phone","address","age","create_time","update_time"],
                        "splitPk": "id",
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false"],
                                "table": ["user_info"]
                            }
                        ],
                        "password": "root",
                        "username": "root",
                        "where": "update_time > '${updateTime}'"
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "writeMode": "replace",
                        "column": ["id","name","sex","phone","address","age","create_time","update_time"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false",
                                "table": ["ods_test_mysql_user_info_m"]
                            }
                        ],
                        "username": "root",
                        "password": "root",
                        "preSql": [],
                        "session": [
                            "set session sql_mode='ANSI'"
                        ]
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": "5"
            }
        }
    }
}

Créer un script d'exécution

Créez un script shell run_sync.sh qui accepte des paramètres dynamiques pour exécuter le job.

#!/bin/bash
# Usage : sh run_sync.sh /chemin/vers/le/job.json 10
jsonPath=$1
echo "Fichier de configuration: $jsonPath"
interval=$2
echo "Intervalle (minutes): $interval"
currentTime=$(date '+%Y-%m-%d %H:%M:%S')
echo "Heure actuelle: $currentTime"
# Calcul de l'heure de début de la fenêtre de synchronisation (pour macOS, adaptez pour Linux)
startTime=$(date -v -${interval}M '+%Y-%m-%d %H:%M:%S')
# Pour Linux, utilisez : startTime=$(date -d "$currentTime $interval minute ago" '+%Y-%m-%d %H:%M:%S')
echo "Fenêtre de début: $startTime"
# Lancer DataX
python3 /chemin/vers/datax/bin/datax.py $jsonPath -p "-DupdateTime='${startTime}'"

Pour exécuter le job en synchronisant les 10 dernières minutes :

./run_sync.sh sync_user_info.json 10

Tester

Exécutez la commande ci-dessus. Le premier lancement peut prendre environ 10 secondes (temps d'initialisation). Cependant, pour des volumes de données importants, les performances sont bien meilleures. Par exemple, la synchronisation de 63 102 enregistrements peut être réalisée en 22 secondes en production.

Utilisation recommandée

L'exemple ci-dessus convient pour une synchronisation ponctuelle. Pour des synchronisations périodiquse, voici des approches plus adaptées :

• Planification via crontab : Solution simple avec crontab -e pour ajouter le script.
• Orchestrateur DolphinScheduler : Très utilisé dans le Big Data (ex: syncrhonisation MySQL vers entrepôt). Il intègre DataX nativement avec une interface graphique, des logs et un suivi.
• Frameworks de tâches planifiées (ElasticJob / XXL-Job) : Ces frameworks métier peuvent exécuter des tâches shell via des paramètres dynamiques.