CDC de DynamoDB vers ClickHouse

ClickPipes vous permet de connecter ClickHouse à DynamoDB.

Cette page explique comment configurer la CDC de DynamoDB vers ClickHouse à l’aide de ClickPipes. Cette intégration comporte 2 composants :

1. L’instantané initial via S3 ClickPipes
2. Les mises à jour en temps réel via Kinesis ClickPipes

Les données seront ingérées dans un ReplacingMergeTree. Ce moteur de table est couramment utilisé dans les scénarios de CDC afin de permettre l’application des opérations de mise à jour. Pour en savoir plus sur cette approche, consultez les articles de blog suivants :

• Change Data Capture (CDC) avec PostgreSQL et ClickHouse - Partie 1
• Change Data Capture (CDC) avec PostgreSQL et ClickHouse - Partie 2

​

1. Configurer le flux Kinesis

Commencez par activer un flux Kinesis sur votre table DynamoDB afin de capturer les modifications en temps réel. Il est préférable de le faire avant de créer le snapshot afin d’éviter toute perte de données. Consultez le guide AWS disponible ici.

​

2. Créer le instantané

Ensuite, nous allons créer un instantané de la table DynamoDB. Cela peut se faire en exportant les données d’AWS vers S3. Consultez le guide AWS disponible ici. Vous devrez effectuer une « Full export » au format DynamoDB JSON.

​

3. Charger le instantané dans ClickHouse

​

Créer les tables nécessaires

Les données du instantané DynamoDB ressembleront à ceci :

{
  "age": {
    "N": "26"
  },
  "first_name": {
    "S": "sally"
  },
  "id": {
    "S": "0A556908-F72B-4BE6-9048-9E60715358D4"
  }
}

Notez que les données sont dans un format imbriqué. Nous devrons les aplatir avant de les charger dans ClickHouse. Pour cela, vous pouvez utiliser la fonction JSONExtract de ClickHouse dans une vue matérialisée.

Nous allons créer trois tables :

1. Une table pour stocker les données brutes de DynamoDB
2. Une table pour stocker les données finales aplaties (table de destination)
3. Une vue matérialisée pour aplatir les données

Pour l’exemple de données DynamoDB ci-dessus, les tables ClickHouse se présenteraient comme suit :

/* Snapshot table */
CREATE TABLE IF NOT EXISTS "default"."snapshot"
(
    `item` String
)
ORDER BY tuple();

/* Table for final flattened data */
CREATE MATERIALIZED VIEW IF NOT EXISTS "default"."snapshot_mv" TO "default"."destination" AS
SELECT
    JSONExtractString(item, 'id', 'S') AS id,
    JSONExtractInt(item, 'age', 'N') AS age,
    JSONExtractString(item, 'first_name', 'S') AS first_name
FROM "default"."snapshot";

/* Table for final flattened data */
CREATE TABLE IF NOT EXISTS "default"."destination" (
    "id" String,
    "first_name" String,
    "age" Int8,
    "version" Int64
)
ENGINE ReplacingMergeTree("version")
ORDER BY id;

La table de destination doit respecter quelques exigences :

• Cette table doit être une table ReplacingMergeTree
• La table doit comporter une colonne version
  • Dans les étapes suivantes, nous ferons correspondre le champ ApproximateCreationDateTime du flux Kinesis à la colonne version.
• La table doit utiliser la clé de partitionnement comme clé de tri (spécifiée par ORDER BY)
  • Les lignes ayant la même clé de tri seront dédupliquées en fonction de la colonne version.

​

Créer le ClickPipe d’instantané

Vous pouvez maintenant créer un ClickPipe pour charger les données de l’instantané de S3 vers ClickHouse. Suivez le guide du S3 ClickPipe ici, mais utilisez les paramètres suivants :

• Chemin d’ingestion : vous devrez repérer le chemin des fichiers JSON exportés dans S3. Le chemin ressemblera à ceci :

https://{bucket}.s3.amazonaws.com/{prefix}/AWSDynamoDB/{export-id}/data/*

• Format: JSONEachRow
• Table: Votre table de instantané (par ex. default.snapshot dans l’exemple ci-dessus)

Une fois la création terminée, les données commenceront à alimenter les tables de instantané et de destination. Vous n’avez pas besoin d’attendre la fin du chargement du instantané avant de passer à l’étape suivante.

​

4. Créer le ClickPipe Kinesis

Nous pouvons maintenant configurer le ClickPipe Kinesis pour capturer les changements en temps réel du flux Kinesis. Suivez le guide ClickPipe Kinesis ici, mais utilisez les paramètres suivants :

• Stream : le flux Kinesis utilisé à l’étape 1
• Table : votre table de destination (par ex. default.destination dans l’exemple ci-dessus)
• Flatten object : true
• Column mappings :
  • ApproximateCreationDateTime : version
  • Mappez les autres champs aux colonnes de destination appropriées, comme indiqué ci-dessous

​

5. Nettoyage (facultatif)

Une fois le ClickPipe de instantané terminé, vous pouvez supprimer la table de l’instantané et la vue matérialisée.

DROP TABLE IF EXISTS "default"."snapshot";
DROP TABLE IF EXISTS "default"."snapshot_clickpipes_error";
DROP VIEW IF EXISTS "default"."snapshot_mv";