> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-fbfa8bee.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
> ClickPipes vous permet de connecter ClickHouse à DynamoDB.
# CDC de DynamoDB vers ClickHouse
export const Image = ({img, alt, size}) => {
return
;
};
export const CloudNotSupportedBadge = () => {
return
Non pris en charge par ClickHouse Cloud
;
};
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](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-1?loc=docs-rockest-migrations)
* [Change Data Capture (CDC) avec PostgreSQL et ClickHouse - Partie 2](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2?loc=docs-rockest-migrations)
## 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](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/kds.html).
## 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](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.HowItWorks.html).
**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 :
```json theme={null}
{
"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 :
```sql theme={null}
/* 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](/fr/integrations/clickpipes/object-storage/amazon-s3/overview), 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 :
```text theme={null}
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](/fr/integrations/clickpipes/kinesis/overview), 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.
```sql theme={null}
DROP TABLE IF EXISTS "default"."snapshot";
DROP TABLE IF EXISTS "default"."snapshot_clickpipes_error";
DROP VIEW IF EXISTS "default"."snapshot_mv";
```