Implement real-time ingestion - Realtime Compute for Apache Flink

Cette rubrique présente les bonnes pratiques pour diffuser en continu les modifications en temps réel depuis MySQL ou Kafka vers un entrepôt de données Hologres à l'aide d'une tâche d'ingestion de données Change Data Capture (CDC) YAML.

Synchroniser MySQL vers Hologres

Le connecteur Hologres prend en charge :

La création automatique de tables et la synchronisation du schéma
Le stockage hybride ligne-colonne et les lectures incrémentielles basées sur le Binlog
Un mappage de types tolérant qui gère les modifications de schéma en amont sans provoquer l'échec des tâches

Le code YAML suivant définit la configuration minimale requise pour synchroniser une base de données MySQL entière avec Hologres :

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: holo_test\.*
  server-id: 8601-8604

sink:
  type: hologres
  name: Hologres Sink
  endpoint: ****.hologres.aliyuncs.com:80
  dbname: cdcyaml_test
  username: ${secret_values.holo-username}
  password: ${secret_values.holo-password}
  sink.type-normalize-strategy: BROADEN

Paramètres optionnels à configurer lors de la première exécution de la tâche :

Paramètre	Objectif	Référence
`include-comments.enabled: true`	Synchroniser les commentaires de table et de champ	Connecteur MySQL
`scan.incremental.snapshot.unbounded-chunk-first.enabled: true`	Donner la priorité aux shards non limités pour éviter les erreurs OutOfMemory de TaskManager	Connecteur MySQL
`scan.only.deserialize.captured.tables.changelog.enabled: true`	Activer les filtres d'analyse pour accélérer la lecture des données	Connecteur MySQL

Gérer les modifications de type en amont

Le connecteur Hologres ne prend pas en charge les événements de modification de type de colonne. Pour gérer les modifications de schéma en amont sans provoquer l'échec des tâches, configurez sink.type-normalize-strategy afin de mapper les types MySQL vers des types Hologres plus larges. La valeur par défaut est STANDARD.

Stratégie	Comportement
`STANDARD`	Mappage de type par défaut
`BROADEN`	Mappage vers des types compatibles plus larges
`ONLY_BIGINT_OR_TEXT`	Mappage de tous les types vers `int8` ou `text`

Par exemple, avec ONLY_BIGINT_OR_TEXT, une modification de type de colonne de INT vers BIGINT dans MySQL mappe les deux types vers int8 dans Hologres ; la tâche se poursuit sans signaler d'erreur de conversion de type.

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: holo_test\.*
  server-id: 8601-8604
  # (Optional) Synchronize table comments and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded shards to prevent potential TaskManager OutOfMemory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: ****.hologres.aliyuncs.com:80
  dbname: cdcyaml_test
  username: ${secret_values.holo-username}
  password: ${secret_values.holo-password}
  sink.type-normalize-strategy: ONLY_BIGINT_OR_TEXT

Pour obtenir la liste complète des mappages de types pris en charge, consultez la section Mappage de type du connecteur Hologres pour les tâches d'ingestion de données YAML.

Écrire dans des tables partitionnées

Lorsque vous utilisez le connecteur Hologres comme destination, vous pouvez écrire des données dans des tables partitionnées. Pour plus d'informations, consultez la section Écrire des données dans des tables partitionnées.

Synchroniser Kafka vers Hologres

Cette section suppose que les modifications MySQL sont déjà acheminées vers Kafka, par exemple en utilisant la configuration décrite dans la section Implémenter la distribution en temps réel. Le pipeline lit les données depuis Kafka et les écrit dans Hologres.

Supposons qu'un topic Kafka nommé inventory contienne des données provenant de deux tables, customers et products, au format debezium-json. La tâche suivante synchronise les deux tables vers leurs tables correspondantes dans Hologres :

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: ****.hologres.aliyuncs.com:80
  dbname: cdcyaml_test
  username: ${secret_values.holo-username}
  password: ${secret_values.holo-password}
  sink.type-normalize-strategy: ONLY_BIGINT_OR_TEXT

transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Formats pris en charge

La source Kafka prend en charge trois formats de message :

Format	Notes
`json`	Format JSON standard
`canal-json`	Format JSON compatible avec Canal
`debezium-json`	Format par défaut pour les données CDC provenant de Debezium

Exigence de clé primaire pour debezium-json

Lorsque le format est debezium-json, la source Kafka ne déduit pas automatiquement les clés primaires. Ajoutez explicitement une clé primaire à l'aide d'une règle de transformation :

transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Tables distribuées

Si les données d'une seule table s'étendent sur plusieurs partitions, ou si des tables fragmentées situées dans différentes partitions doivent être fusionnées, définissez debezium-json.distributed-tables ou canal-json.distributed-tables sur true.

Inférence de schéma

La source Kafka prend en charge plusieurs stratégies d'inférence de schéma. Configurez la stratégie à l'aide du paramètre schema.inference.strategy. Pour plus de détails, consultez la section Message Queue for Apache Kafka.