DataWorks:Configurer un nœud de synchronisation par lot pour les données incrémentielles

Prenons l'exemple de la synchronisation des données MySQL :

• Si vous ne configurez pas Data Filtering, toutes les données sont synchronisées vers la table de destination par défaut.

• Si vous configurez Data Filtering, seules les données répondant à la condition de filtrage sont synchronisées vers la table de destination.

Un paramètre de planification spécifie le nom de partition de la table MaxCompute de destination. $bizdate représente la date métier. Lorsqu'une tâche planifiée s'exécute, le système remplace l'expression de partition configurée pour la tâche par la date métier représentée par le paramètre de planification. Pour des instructions de configuration détaillées sur les expressions des paramètres de planification, consultez Scénarios d'application des paramètres de planification dans Data Integration. Prenons l'exemple d'une tâche de synchronisation par lot. Vous devez configurer le paramètre bizdate à trois endroits pour mettre en œuvre la synchronisation incrémentielle : Dans la section Data Filtering de la source, saisissez STR_TO_DATE('${bizdate}','%Y%m%d') <= gmt_modify_time AND gmt_modify_time < DATE_ADD(STR_TO_DATE('${bizdate}','%Y%m%d'), interval 1 day) pour filtrer les données modifiées à la date métier. Dans la section Partition Information de la destination, saisissez pt=${bizdate} pour écrire les données dans la partition de date correspondante, et définissez la Cleanup Rule sur Clean up existing data before writing (Insert Overwrite). Dans la section Parameters des Schedule Settings sur le côté droit, saisissez bizdate=$bizdate afin que le système de planification remplace automatiquement ${bizdate} par la date métier réelle lors de l'exécution. Lors de la configuration de la synchronisation incrémentielle des données :

• Synchronisation incrémentielle basée sur des colonnes de type temps : Utilisez des paramètres de planification pour remplacer dynamiquement les données de type temps. Lors de la planification des tâches, les paramètres de planification sont automatiquement remplacés par des valeurs spécifiques en fonction de la date métier. Pour plus d'informations sur les paramètres de planification, consultez Configurer les paramètres de planification.

• Synchronisation incrémentielle basée sur des colonnes non temporelles : Utilisez un nœud d'affectation pour convertir la colonne vers le type de données cible, puis transmettez-la à Data Integration pour la synchronisation. Pour plus d'informations sur les nœuds d'affectation, consultez Créer un nœud d'affectation.

Remarques

Lorsque vous configurez une tâche de synchronisation incrémentielle, tenez compte des points suivants :

• Sécurité du nettoyage des données existantes avant l'écriture (Insert Overwrite) : Lorsque plusieurs tâches de synchronisation écrivent dans différentes partitions de la même table MaxCompute, la stratégie Insert Overwrite est sûre. Cette stratégie efface uniquement les données de partition spécifiées par la tâche actuelle et n'affecte pas les données des autres partitions de la table, empêchant ainsi les conflits de données ou les suppressions accidentelles.

• Limitation de l'écrasement par lot de la plage de partitions : DataWorks ne prend pas en charge la spécification d'une plage horaire (telle que hh=00-23) dans la configuration de partition pour l'écrasement par lot. Pour écraser les données de plusieurs heures, configurez des tâches distinctes pour chaque heure. Les paramètres de partition ne prennent actuellement en charge qu'une seule valeur spécifique ou le caractère générique *.

• Syntaxe des caractères génériques : Lorsque la source contient une partition au niveau de l'heure mais que la destination ne possède qu'une partition au niveau du jour, saisissez le caractère générique * dans le champ de partition horaire pour faire correspondre toutes les données horaires. Saisissez * directement sans guillemets (tels que "*"). Sinon, une erreur de syntaxe se produit.

Synchronisation incrémentielle planifiée à haute fréquence basée sur l'horodatage

DataWorks prend en charge la synchronisation incrémentielle planifiée basée sur l'horodatage en combinant des tâches de synchronisation par lot avec une planification périodique (par exemple, toutes les 5 minutes ou toutes les heures). Cette approche convient aux scénarios de synchronisation T+1 ou quasi en temps réel depuis RDS MySQL vers des destinations telles que SelectDB et StarRocks. Elle met en œuvre la synchronisation incrémentielle via le filtrage SQL sans nécessiter de tâches CDC en temps réel, ce qui évite les coûts liés à l'exécution continue des tâches. Points de configuration clés :

1. Dans la condition where de Data Filtering de la source, utilisez une colonne d'horodatage comme variable de filtrage. Par exemple : gmt_modify_time >= '$[yyyymmddhhmiss-10/mi]' AND gmt_modify_time < '$[yyyymmddhhmiss]'.

2. Configurez des paramètres de planification périodiques (tels que $[yyyymmddhhmiss]) pour calculer dynamiquement la plage horaire, garantissant que chaque exécution planifiée ne synchronise que les données incrémentielles dans l'intervalle de temps spécifié. Pour la configuration détaillée des paramètres de planification, consultez Scénarios d'application des paramètres de planification dans Data Integration.

3. Dans le mappage de colonnes de destination, ajoutez manuellement un paramètre constant mappé à la colonne de partition pour activer l'écriture dynamique des partitions.

Configuration incrémentielle pour les tâches de synchronisation par lot au niveau de la base de données

Outre les tâches de synchronisation par table unique, vous pouvez également créer une tâche de synchronisation par lot au niveau de la base de données pour mettre en œuvre une synchronisation incrémentielle périodique. Lors de la création de la tâche, sélectionnez la synchronisation incrémentielle et configurez la condition incrémentielle dans la tâche de synchronisation au niveau de la base de données. Cela permet une synchronisation incrémentielle efficace des partitions au niveau du jour (par exemple, en filtrant par la colonne create_time).

Cette approche convient aux scénarios où vous souhaitez gérer de manière centralisée la synchronisation pour plusieurs tables, mais n'avez besoin d'un traitement incrémentiel que pour des tables spécifiques. Pour le processus de configuration complet des tâches de synchronisation par lot au niveau de la base de données, consultez Configurer une tâche de synchronisation par lot au niveau de la base de données.

Que faire si une erreur « partition not found » se produit après avoir utilisé des paramètres de planification dans le filtre de partition du lecteur MaxCompute ?

Cause : Les paramètres de planification configurés ne sont pas correctement résolus en valeurs de partition réelles lors de l'exécution, ou les valeurs résolues ne correspondent pas aux partitions réelles de la table source.

Solution : Si la valeur de partition est transmise à partir du paramètre outputs d'un nœud en amont, vérifiez la configuration des paramètres dans Data Studio et assurez-vous que les conditions suivantes sont remplies :

• Le nom du paramètre configuré est cohérent avec le nom du paramètre d'entrée.

• La valeur du paramètre transmis correspond exactement à la partition réelle dans MaxCompute.

Une tâche de synchronisation par lot DataWorks effectue-t-elle une synchronisation complète ou incrémentielle par défaut ? Comment configurer la synchronisation incrémentielle pour une table source sans colonnes de partition ?

Comportement par défaut

Les tâches de synchronisation par lot Data Integration de DataWorks effectuent une synchronisation complète par défaut, ce qui signifie que toutes les données sont synchronisées à chaque fois. La synchronisation incrémentielle n'est activée que lorsque vous configurez une condition Data Filtering combinée avec des paramètres de planification.

Gestion des tables sources sans colonnes de partition

Si la table de la base de données source (telle qu'une table RDS) ne possède pas de colonne temporelle ou de partition, vous ne pouvez pas filtrer directement les données incrémentielles à l'aide d'une condition where. Nous vous recommandons d'ajouter une colonne temporelle (telle que dt ou gmt_modify_time) à la table source comme base pour le filtrage incrémentiel. Une fois la colonne prête, configurez la logique de synchronisation incrémentielle en vous référant à la section « Configurer la synchronisation incrémentielle » de cette rubrique.

Concepts clés

Définitions des paramètres

• splitPk (clé de division) : Spécifie une colonne de clé primaire. DataWorks divise les données en plusieurs lots en fonction de la plage de valeurs de cette colonne pour permettre des lectures simultanées multithread.

• splitFactor (facteur de division) : Contrôle la granularité de la division. Une valeur plus élevée entraîne des divisions plus fines et davantage de threads de lecture.

Impact sur les performances et recommandations

L'activation de splitPk et splitFactor augmente la charge sur la base de données source. Pour réduire la charge sur la source, nous recommandons :

• Réduisez la concurrence, ou définissez la concurrence de la tâche de synchronisation par lot au niveau de la base de données sur 1.

• Assurez-vous que la colonne de division (splitPk) est indexée pour améliorer l'efficacité de la lecture.