DataWorks:Source de données PolarDB

Lecture et écriture par lots

La lecture des données depuis des vues est possible.

Lecture en temps réel

Si vous utilisez un cluster PolarDB for MySQL comme source, activez le journal binaire. PolarDB for MySQL est une base de données cloud-native entièrement compatible avec MySQL. Par défaut, PolarDB for MySQL utilise des journaux physiques de haut niveau plutôt que le journal binaire. Pour une meilleure intégration à l'écosystème MySQL, PolarDB permet d'activer le journal binaire.

Lecture par lots

Le tableau suivant présente les mappages de types de données pour PolarDB Reader.

Écriture par lots

Comme PolarDB Reader, PolarDB Writer prend en charge la plupart des types de données PolarDB, mais pas tous. Vérifiez la compatibilité de vos types de données.

Le tableau suivant présente les mappages de types de données pour PolarDB Writer.

Configurer une liste blanche d'adresses IP

Ajoutez le bloc CIDR du VPC contenant votre groupe de ressources Serverless ou groupe de ressources exclusif pour Data Integration à la liste blanche d'adresses IP de votre cluster PolarDB. Pour plus d'informations, consultez la rubrique Définir une liste blanche de cluster.

Créer un compte et accorder des autorisations

Créez un compte et accordez-lui les autorisations requises.

Créez un compte de base de données dédié à la synchronisation des données. Ce compte doit disposer des autorisations SELECT, REPLICATION SLAVE, REPLICATION CLIENT sur la base de données.

1. Créez un compte.

   Pour plus d'informations, consultez la rubrique Créer et gérer un compte de base de données.

2. Accordez des autorisations.

   Exécutez la commande suivante pour accorder les autorisations requises au compte. Vous pouvez également accorder l'autorisation SUPER.

   -- CREATE USER 'sync_account'@'%' IDENTIFIED BY 'your_password';
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'sync_account'@'%';

Activer la journalisation binaire

Pour plus d'informations, consultez la rubrique Activer la journalisation binaire.

Configurer une tâche par lots pour une seule table

• Pour plus d'informations, consultez les rubriques Configurer une tâche dans l'interface sans code et Configurer une tâche en mode script.

• Pour obtenir la liste complète des paramètres et un exemple de script, consultez la section Annexe : Exemple de script et paramètres.

Configurer la synchronisation en temps réel pour une table ou une base de données

Pour plus d'informations, consultez la rubrique Configurer une tâche de synchronisation en temps réel (héritée).

Configurer la lecture par lots de toute la base de données et la synchronisation incrémentielle en temps réel

Pour plus d'informations, consultez la rubrique Configurer une tâche de synchronisation en temps réel pour toute la base de données.

Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code

Pour configurer une tâche de synchronisation par lots via l'éditeur de code, définissez les paramètres associés dans le script conformément aux exigences de format unifié. Pour plus d'informations, consultez la rubrique Configuration en mode script. Les informations suivantes décrivent les paramètres à configurer pour les sources de données lors de la configuration d'une tâche de synchronisation par lots via l'éditeur de code.

Exemple de script Reader

Le code suivant fournit un exemple de script pour la lecture des données d'une seule table dans une base de données. Pour plus d'informations sur les paramètres, consultez la section de description des paramètres.

{
    "type": "job",
    "steps": [
        {
            "parameter": {
                "datasource": "test_005",                // The name of the data source.
                "column": [                              // The source column names.
                    "id",
                    "name",
                    "age",
                    "sex",
                    "salary",
                    "interest"
                ],
                "where": "id=1001",                      // The filter condition.
                "splitPk": "id",                         // The sharding key.
                "table": "PolarDB_person",               // The source table name.
              	"useReadonly": "false"                   // Specifies whether to read data from a secondary database.
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {}
    ],
    "version": "2.0",                                // The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {                              // The error tolerance.
            "record": ""
        },
        "speed": {
            "concurrent": 6,                         // The concurrency level.
            "throttle": true,                        // If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
      "mbps":"12"                              // The throttling rate, in MB/s.
        }
    }
}

Paramètres du script Reader

Exemple de script Writer

Le code suivant fournit un exemple de configuration de script. Pour plus d'informations sur les paramètres, consultez la section de description des paramètres.

{
    "type": "job",
    "steps": [
        {
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {
                "postSql": [],                        // The SQL statement to be executed after the synchronization task is complete.
                "datasource": "test_005",             // The name of the data source.
                "column": [                           // The destination column names.
                    "id",
                    "name",
                    "age",
                    "sex",
                    "salary",
                    "interest"
                ],
                "writeMode": "insert",                // The write mode.
                "batchSize": 256,                     // The number of records to submit in each batch.
                "table": "PolarDB_person_copy",       // The destination table name.
                "preSql": []                          // The SQL statement to be executed before the synchronization task starts.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",                             // The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {                           // The error tolerance.
            "record": ""
        },
        "speed": {
            "throttle":true,                      // If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
            "concurrent":6,                       // The concurrency level.
            "mbps":"12"                           // The throttling rate, in MB/s.
        }
    }
}

Paramètres du script Writer

• Tous les paramètres

  Paramètre	Description	Obligatoire	Valeur par défaut
  datasource	Nom de la source de données. En mode script, la valeur de ce paramètre doit correspondre exactement au nom de la source de données ajoutée.	Oui	Aucune
  table	Nom de la table de destination vers laquelle vous souhaitez synchroniser les données.	Oui	Aucune
  writeMode	Mode d'écriture. Valeurs valides : insert : correspond à INSERT INTO dans l'interface sans code. update : correspond à ON DUPLICATE KEY UPDATE dans l'interface sans code. replace : correspond à REPLACE INTO dans l'interface sans code. Pour plus d'informations sur les modes et des exemples, consultez la section Détails du paramètre writeMode ci-dessous. Remarque Pour PolarDB for PostgreSQL, seul le mode insert est pris en charge. Pour mettre à jour les données et éviter les conflits de clé primaire, supprimez les données en double avant d'exécuter la tâche de synchronisation par lots. Les méthodes suivantes sont recommandées : Méthode 1 : Dans le paramètre preSql (qui correspond à Pre-Import Statement dans l'interface sans code), configurez une instruction TRUNCATE pour vider la table de destination. Méthode 2 : Traitez la table de destination dans un nœud en amont pour éviter les conflits de clé primaire lors de la synchronisation des données.	Non	insert
  column	Colonnes de destination vers lesquelles vous souhaitez écrire les données. Séparez les colonnes par des virgules (,). Exemple : "column": ["id", "name", "age"]. Pour écrire les données dans toutes les colonnes dans l'ordre, utilisez un astérisque (). Exemple : "column": [""].	Oui	Aucune
  preSql	Spécifie une ou plusieurs instructions SQL à exécuter avant le démarrage de la tâche. L'interface sans code prend en charge une seule instruction, tandis que le mode script prend en charge plusieurs instructions, telles que des instructions pour effacer les données existantes.	Non	Aucune
  postSql	Spécifie une ou plusieurs instructions SQL à exécuter une fois la tâche terminée. L'interface sans code prend en charge une seule instruction, tandis que le mode script prend en charge plusieurs instructions, telles qu'une instruction pour ajouter un horodatage.	Non	Aucune
  batchSize	Nombre d'enregistrements à soumettre par lot. Des valeurs plus élevées peuvent améliorer le débit en réduisant les interactions réseau avec PolarDB, mais une valeur excessivement élevée peut provoquer une erreur de mémoire insuffisante (OOM).	Non	1024
  updateColumn	Colonnes à mettre à jour en cas de conflit de clé primaire ou d'index unique. Ce paramètre n'est valide que lorsque writeMode est défini sur update. Vous pouvez spécifier plusieurs colonnes, séparées par des virgules. Exemple : "updateColumn": ["name", "age"]. Remarque Ce paramètre est pris en charge uniquement pour PolarDB for MySQL.	Non	Aucune

• Détails du paramètre writeMode

  Comparaison	insert (correspond à INSERT INTO dans l'interface sans code)	update (correspond à ON DUPLICATE KEY UPDATE dans l'interface sans code)	replace (correspond à REPLACE INTO dans l'interface sans code)
  Stratégie de gestion des conflits	En cas de conflit de clé primaire ou d'index unique, la ligne conflictuelle n'est pas écrite dans la table de destination et est traitée comme des données erronées.	S'il n'y a aucun conflit de clé primaire ou d'index unique, ce mode fonctionne de la même manière que le mode INSERT INTO. En cas de conflit, la nouvelle ligne met à jour uniquement les champs spécifiés de la ligne existante.	S'il n'y a aucun conflit de clé primaire ou d'index unique, ce mode fonctionne de la même manière que le mode INSERT INTO. En cas de conflit, la ligne existante est supprimée et la nouvelle ligne est insérée, remplaçant ainsi tous ses champs.
  Exemple de données	Table source +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Table de destination d'origine +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | | +----+---------+-----+ Une fois la tâche exécutée, une ligne est écrite dans la table de destination et une ligne est enregistrée comme donnée erronée. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | wangwu | | +----+---------+-----+	Scénario 1 : La tâche est configurée pour synchroniser uniquement certaines colonnes : "column": ["id","name"] Table source +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Table de destination d'origine +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ Une fois la tâche exécutée, deux lignes sont écrites dans la table de destination et aucune donnée erronée n'est enregistrée. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | 3 | +----+---------+-----+ Scénario 2 : La tâche est configurée pour synchroniser toutes les colonnes : "column": ["id","name","age"] Table source +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Table de destination d'origine +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ Une fois la tâche exécutée, deux lignes sont écrites dans la table de destination et aucune donnée erronée n'est enregistrée. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+	Table source +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ Table de destination d'origine +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ Une fois la tâche exécutée, deux lignes sont écrites dans la table de destination et aucune donnée erronée n'est enregistrée. +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+