Architecture et Principe de DataX

DataX est une plateforme de synchronisation de données hors ligne open-source, initialement développée pour répondre aux besoins internes d'Alibaba. Elle permet de transférer des données entre diverses sources hétérogènes. Le framework repose sur une architecture modulaire où chaque source de données est gérée par un plugin Reader (pour l'extraction) et un plugin Writer (pour le chargement). Cette conception extensible garantit que l'ajout d'un nouveau connecteur le rend immédiatement compatible avec l'ensemble de l'écosystème existant.

Préparation de l'Environnement Cible

Avant d'initier le transfert, il est nécessaire de préparer les structures de destination dans TDengine. Cette étape peut être réalisée via l'interface en ligne de commande du conteneur ou en utilisant des clients graphiques comme DBeaver ou DataGrip pour créer les super-tables et les sous-tables requises.

Configuration du Job de Synchronisation

Le cœur de la configuration réside dans le fichier JSON du job. Voici un exemple de configuration pour extraire des métriques de capteurs depuis MySQL et les charger dans TDengine :

{
  "job": {
    "setting": {
      "speed": {
        "channel": 2
      }
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "db_admin",
            "password": "SecurePass123!",
            "connection": [
              {
                "querySql": [
                  "SELECT ts, temp, humidity, pressure, wind_spd, wind_dir FROM iot_sensor_metrics;"
                ],
                "jdbcUrl": [
                  "jdbc:mysql://192.168.1.100:3306/iot_db?useUnicode=true&characterEncoding=utf8&useSSL=false"
                ]
              }
            ]
          }
        },
        "writer": {
          "name": "tdengine30writer",
          "parameter": {
            "username": "root",
            "password": "taosdata",
            "column": [
              "ts", "temp", "humidity", "pressure", "wind_spd", "wind_dir"
            ],
            "connection": [
              {
                "table": ["iot_sensor_metrics"],
                "jdbcUrl": "jdbc:TAOS-RS://192.168.1.200:6041/iot_db"
              }
            ],
            "batchSize": 500,
            "ignoreTagsUnmatched": true
          }
        }
      }
    ]
  }
}

Gestion de la Compatibilité TDengine 3.x

Une difficulté fréquente lors de l'utilisation de TDengine version 3.x (comme la 3.2.3.0) est l'incompatibilité syntaxique avec l'ancien plugin tdenginewriter inclus par défaut, qui ne supporte que la série 2.x. Pour résoudre ce problème, il faut récupérer les sources du plugin tdengine30writer depuis le dépôt officiel, le compiler, et intégrer le package résultant dans le répertoire plugin/writer de votre installation DataX. N'oubliez pas de mettre à jour le fichier plugin.json correspondant.

Une fois configuré, le job s'exécute via la commande :

python datax.py ./job/iot_migration_job.json

Initialisation des Services TDengine

Selon le mode de connexion choisi, différents services doivent être actifs sur le serveur cible :

• Connexion Native (Port 6030) : Nécessite le démon principal. L'URL JDBC sera jdbc:TAOS://IP_SERVEUR:6030/iot_db.

systemctl start taosd

• Connexion REST (Port 6041) : Nécessite l'adaptateur REST. L'URL JDBC sera jdbc:TAOS-RS://IP_SERVEUR:6041/iot_db.

systemctl start taosadapter

Implémentation Java Alternative pour la Migration

Dans certains cas, un contrôle plus fin du processus de migration est requis. Voici une implémentation Java refactorisée qui lit les identifiants d'appareils depuis MySQL, construit les requêtes d'insertion, et pousse les données vers TDengine en utilisant la syntaxe de création automatique de sous-tables.

package com.example.datasync;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class MetricsDataMigrator {

    private static final String MYSQL_URL = "jdbc:mysql://192.168.1.100:3306/iot_db?useSSL=false&characterEncoding=UTF-8";
    private static final String MYSQL_USER = "db_admin";
    private static final String MYSQL_PASS = "SecurePass123!";

    private static final String TDENGINE_URL = "jdbc:TAOS-RS://192.168.1.200:6041/iot_db";
    private static final String TDENGINE_USER = "root";
    private static final String TDENGINE_PASS = "taosdata";

    public static void main(String[] args) {
        List<String> deviceIds = fetchDeviceIds();
        for (String deviceId : deviceIds) {
            migrateDeviceMetrics(deviceId);
        }
        System.out.println("Processus de migration terminé.");
    }

    private static List<String> fetchDeviceIds() {
        List<String> ids = new ArrayList<>();
        String query = "SELECT DISTINCT device_id FROM iot_sensor_metrics ORDER BY device_id";
        
        try (Connection conn = DriverManager.getConnection(MYSQL_URL, MYSQL_USER, MYSQL_PASS);
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(query)) {
            while (rs.next()) {
                ids.add(rs.getString("device_id"));
            }
        } catch (SQLException e) {
            System.err.println("Erreur lors de la récupération des IDs : " + e.getMessage());
        }
        return ids;
    }

    private static void migrateDeviceMetrics(String deviceId) {
        String selectQuery = "SELECT ts, temp, humidity, pressure, wind_spd, wind_dir FROM iot_sensor_metrics WHERE device_id = ?";
        StringBuilder valuesBuilder = new StringBuilder();

        try (Connection mysqlConn = DriverManager.getConnection(MYSQL_URL, MYSQL_USER, MYSQL_PASS);
             PreparedStatement pstmt = mysqlConn.prepareStatement(selectQuery)) {

            pstmt.setString(1, deviceId);
            try (ResultSet rs = pstmt.executeQuery()) {
                while (rs.next()) {
                    valuesBuilder.append(String.format("('%s', %f, %f, %f, %f, '%s') ",
                            rs.getTimestamp("ts"),
                            rs.getDouble("temp"),
                            rs.getDouble("humidity"),
                            rs.getDouble("pressure"),
                            rs.getDouble("wind_spd"),
                            rs.getString("wind_dir")));
                }
            }
        } catch (SQLException e) {
            System.err.println("Erreur de lecture MySQL pour l'appareil " + deviceId + " : " + e.getMessage());
            return;
        }

        if (valuesBuilder.length() > 0) {
            pushToTdengine(deviceId, valuesBuilder.toString());
        }
    }

    private static void pushToTdengine(String deviceId, String values) {
        String insertSql = String.format(
                "INSERT INTO sensor_metrics_%s USING iot_sensor_metrics TAGS('%s', 'system') VALUES %s",
                deviceId, deviceId, values);

        try (Connection taosConn = DriverManager.getConnection(TDENGINE_URL, TDENGINE_USER, TDENGINE_PASS);
             Statement stmt = taosConn.createStatement()) {
            stmt.executeUpdate(insertSql);
            System.out.println("Données intégrées pour l'appareil : " + deviceId);
        } catch (SQLException e) {
            System.err.println("Erreur d'écriture TDengine pour l'appareil " + deviceId + " : " + e.getMessage());
        }
    }
}