Data synchronization capabilities of the DataWorks AnalyticDB for PostgreSQL connector - DataWorks

Limites

La synchronisation par lots prend en charge la lecture des données depuis une vue.

Versions prises en charge

Le connecteur prend en charge les versions jusqu'à la 7.0 incluse.

Types de données pris en charge

Lecture par lots

AnalyticDB for PostgreSQL Reader prend en charge la plupart des types de données AnalyticDB for PostgreSQL. Vérifiez que vos types de données sont pris en charge avant utilisation.

Le tableau suivant répertorie les types de données pris en charge pour les lectures par lots.

Catégorie

Type AnalyticDB for PostgreSQL

Entier

BIGINT, BIGSERIAL, INTEGER, SMALLINT, SERIAL et GEOMETRY

Virgule flottante

DOUBLE, PRECISION, MONEY, NUMERIC et REAL

Chaîne

VARCHAR, CHAR, TEXT, BIT et INET

Date et heure

DATE, TIME et TIMESTAMP

Booléen

BOOL

Binaire

BYTEA

Écriture par lots

AnalyticDB for PostgreSQL Writer prend en charge la plupart des types de données AnalyticDB for PostgreSQL. Vérifiez que vos types de données sont pris en charge avant utilisation.

Le tableau suivant établit la correspondance entre les types de données DataWorks et les types de données AnalyticDB for PostgreSQL pour les écritures par lots.

Type de données DataWorks

Type AnalyticDB for PostgreSQL

LONG

BIGINT, BIGSERIAL, INTEGER, SMALLINT et SERIAL

DOUBLE

DOUBLE, PRECISION, MONEY, NUMERIC et REAL

STRING

VARCHAR, CHAR, TEXT, BIT, INET et GEOMETRY

DATE

DATE, TIME et TIMESTAMP

BOOLEAN

BOOL

BYTES

BYTEA

Remarque

Utilisez un cast de type, tel que a_inet::varchar, pour les types de données MONEY, INET et BIT.

Ajouter une source de données

Avant de développer une tâche de synchronisation dans DataWorks, ajoutez la source de données requise à DataWorks en suivant les instructions fournies dans la rubrique Configuration de la source de données. Consultez les descriptions des paramètres dans la console DataWorks pour comprendre leur signification lors de l'ajout d'une source de données.

Développer une tâche de synchronisation des données

Pour obtenir des informations sur le point d'entrée et la procédure de configuration d'une tâche de synchronisation, consultez les guides de configuration suivants.

Synchronisation par lots d'une seule table

Pour la procédure, consultez les rubriques Configurer une tâche de synchronisation à l'aide de l'interface sans code et Configurer une tâche de synchronisation à l'aide de l'éditeur de code.
Pour obtenir la liste complète des paramètres et des exemples de scripts pour l'éditeur de code, consultez la rubrique Annexe : Exemples de scripts et paramètres.

Synchronisation par lots de toute la base de données

Pour la procédure, consultez la rubrique Configurer une tâche de synchronisation par lots pour une base de données entière.

Annexe : Exemples de scripts et paramètres

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

Pour configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, configurez les paramètres associés dans le script en respectant les exigences de format de script 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 à l'aide de l'éditeur de code.

Exemple de script Reader

{
    "type": "job",
    "steps": [
        {
            "parameter": {
                "datasource": "test_004",// The name of the data source.
                "column": [// The columns in the source table.
                    "id",
                    "name",
                    "sex",
                    "salary",
                    "age"
                ],
                "where": "id=1001",// The WHERE clause.
                "splitPk": "id",// The shard key.
                "table": "public.person"// The name of the source table.
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {},
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",// The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {// The error record count.
            "record": ""
        },
        "speed": {
            "concurrent": 6,// The number of concurrent threads.
            "throttle": true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
           "mbps":"12"// The throttling rate in MB/s.
        }
    }
}

Paramètres Reader

Paramètre	Description	Obligatoire	Valeur par défaut
datasource	Nom de la source de données telle qu'enregistrée dans DataWorks.	Oui	Aucune
table	Nom de la table source.	Oui	Aucune
column	Colonnes à synchroniser, spécifiées sous forme de tableau JSON. Par défaut, toutes les colonnes sont synchronisées, comme indiqué par [*]. Sélectionnez un sous-ensemble de colonnes à exporter. Modifiez l'ordre des colonnes. Configurez des constantes. Respectez la syntaxe SQL. Exemple : `["id", "table","1","'mingya.wmy'","'null'", "to_char(a+1)","2.3","true"]`. id est un nom de colonne standard. table est un nom de colonne qui est également un mot clé réservé. 1 est une constante entière. 'mingya.wmy' est une constante de chaîne. La valeur doit être entourée d'apostrophes ('). 'null' est une constante de chaîne. to_char(a+1) est une expression de fonction. 2.3 est un nombre à virgule flottante. true est une valeur booléenne. Spécifiez explicitement les colonnes à synchroniser pour le paramètre column. La valeur ne peut pas être vide.	Oui	Aucune
splitPk	Clé de fragmentation. Définissez splitPk sur une colonne pour activer l'extraction parallèle des données via le sharding. Nous recommandons d'utiliser la clé primaire de la table, qui est généralement distribuée de manière homogène et permet d'éviter les points chauds de données. Le paramètre splitPk prend en charge le sharding des données uniquement pour les colonnes de type entier. Si vous spécifiez une colonne d'un autre type de données, tel que chaîne, flottant ou date, Data Integration ignore la configuration splitPk et synchronise les données via un seul thread. Si vous ne configurez pas le paramètre splitPk ou si vous le laissez vide, la tâche synchronise les données à l'aide d'un seul thread.	Non	Aucune
where	Clause WHERE utilisée pour filtrer les données extraites. Le lecteur construit une requête SQL à partir des paramètres column, table et where. Exemple : `id>2 and sex=1`. Utilisez la clause WHERE pour effectuer une synchronisation incrémentielle. Si vous ne spécifiez pas de clause WHERE, le connecteur synchronise l'intégralité de la table.	Non	Aucune
querySql (avancé ; non disponible dans l'interface sans code)	Requête SQL personnalisée pour le filtrage des données. Lorsque le paramètre `where` ne permet pas d'exprimer vos conditions de filtre, utilisez ce paramètre à la place. Data Integration ignore les paramètres column, table et where lorsque celui-ci est configuré. Exemple : `select a,b from table_a join table_b on table_a.id = table_b.id`. Lorsque vous configurez le paramètre querySql, AnalyticDB for PostgreSQL Reader ignore les paramètres column, table et where.	Non	Aucune
fetchSize	Nombre d'enregistrements à extraire par lot depuis la base de données. Une valeur plus élevée réduit les allers-retours réseau et peut améliorer les performances d'extraction. Remarque Une valeur fetchSize supérieure à 2048 peut provoquer une erreur de mémoire insuffisante (OOM) lors du processus de synchronisation des données.	Non	512

Exemple de script Writer

{
    "type": "job",
    "steps": [
        {
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "parameter": {
                "postSql": [],// The post-execution SQL statement.
                "datasource": "test_004",// The name of the data source.
                "column": [// The columns in the destination table.
                    "id",
                    "name",
                    "sex",
                    "salary",
                    "age"
                ],
                "table": "public.person",// The name of the destination table.
                "preSql": []// The pre-execution SQL statement.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",// The version number.
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {// The error record count.
            "record": ""
        },
        "speed": {
            "throttle":true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
            "concurrent":6, // The number of concurrent threads.
            "mbps":"12"// The throttling rate.
        }
    }
}

Paramètres Writer

Paramètre	Description	Obligatoire	Valeur par défaut
datasource	Nom de la source de données telle qu'enregistrée dans DataWorks.	Oui	Aucune
table	Nom de la table de destination.	Oui	Aucune
writeMode	Mode d'écriture. Valeurs valides : insert, copy et upsert. insert : Exécute l'instruction PostgreSQL `INSERT INTO ... VALUES ...` pour écrire des données dans PostgreSQL. Ce mode est recommandé. copy : PostgreSQL fournit la commande COPY pour copier des données entre une table et un fichier (entrée standard ou sortie standard). Data Integration prend en charge l'utilisation de `COPY FROM` pour charger des données dans une table. Utilisez ce mode pour améliorer les performances. upsert : En cas de conflit de données lors de l'opération d'écriture, le système traite les nouvelles et les données existantes en fonction du paramètre conflictMode.	Non	insert
conflictMode	Si writeMode est défini sur upsert et qu'un conflit de clé primaire ou d'index unique se produit lors de l'écriture des données dans PostgreSQL, sélectionnez l'une des politiques de gestion des conflits suivantes : replace : En cas de conflit, les nouvelles données entrantes écrasent les données existantes. ignore : En cas de conflit, le connecteur ignore les nouvelles données et conserve les données existantes. Remarque Vous pouvez configurer la politique de gestion des conflits uniquement dans l'éditeur de code.	Non	replace
column	Colonnes de destination vers lesquelles écrire les données, séparées par des virgules (,). Exemple : `"column":["id","name","age"]`. Pour écrire dans toutes les colonnes dans l'ordre, utilisez un astérisque (). Exemple : "column":[""].	Oui	Aucune
preSql	Instruction SQL exécutée avant le démarrage de la tâche de synchronisation. L'interface sans code prend en charge une seule instruction ; l'éditeur de code prend en charge plusieurs instructions, par exemple pour effacer les anciennes données.	Non	Aucune
postSql	Instruction SQL exécutée après la fin de la tâche de synchronisation. L'interface sans code prend en charge une seule instruction ; l'éditeur de code prend en charge plusieurs instructions, par exemple pour ajouter un horodatage.	Non	Aucune
batchSize	Nombre d'enregistrements à écrire par lot. Une valeur plus élevée réduit les allers-retours réseau et améliore le débit, mais une valeur excessivement grande peut provoquer une erreur OOM.	Non	1024