メインコンテンツまでスキップ
メインコンテンツまでスキップ

DynamoDBからClickHouseへのCDC

Experimental feature. Learn more.

このページでは、ClickPipesを使用してDynamoDBからClickHouseへのCDCを設定する方法について説明します。この統合には2つのコンポーネントがあります：

  1. S3 ClickPipesを介した初期スナップショット
  2. Kinesis ClickPipesを介したリアルタイム更新

データは ReplacingMergeTree に取り込まれます。このテーブルエンジンは、更新操作が適用されるCDCシナリオで一般的に使用されます。このパターンに関する詳細は、以下のブログ記事を参照してください：

1. Kinesisストリームの設定​

まず、DynamoDBテーブルでリアルタイムの変更をキャプチャするためにKinesisストリームを有効にします。スナップショットを作成する前にこれを行うことで、データを見逃すことを避けることができます。 AWSガイドはこちらにあります。

DynamoDB Kinesis Stream

2. スナップショットを作成する​

次に、DynamoDBテーブルのスナップショットを作成します。これはAWSエクスポートを使用してS3に達成できます。AWSガイドはこちらにあります。 DynamoDB JSON形式で「フルエクスポート」を行う必要があります。

DynamoDB S3 Export

3. スナップショットをClickHouseにロードする​

必要なテーブルを作成する​

DynamoDBからのスナップショットデータは次のようになります：

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

データがネストされた形式であることに注意してください。このデータをClickHouseにロードする前にフラット化する必要があります。これは、ClickHouseのマテリアライズドビューのJSONExtract関数を使用して行うことができます。

3つのテーブルを作成したいと思います：

  1. DynamoDBからの生データを保存するテーブル
  2. 最終的なフラット化されたデータを保存するテーブル（宛先テーブル）
  3. データをフラット化するためのマテリアライズドビュー

上記のDynamoDBデータの例に対して、ClickHouseのテーブルは次のようになります：

/* 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;

宛先テーブルにはいくつかの要件があります：

  • このテーブルはReplacingMergeTreeテーブルである必要があります
  • テーブルにはversionカラムが必要です
    • 後のステップでは、KinesisストリームのApproximateCreationDateTimeフィールドをversionカラムにマッピングします。
  • テーブルは、ソートキーとしてパーティションキーを使用する必要があります（ORDER BYで指定）
    • 同じソートキーを持つ行はversionカラムに基づいて重複排除されます。

スナップショットClickPipeを作成する​

S3からClickHouseにスナップショットデータをロードするためのClickPipeを作成できます。S3 ClickPipeガイドはこちらに従ってくださいが、次の設定を使用してください：

  • インジェストパス：エクスポートされたjsonファイルのS3内のパスを特定する必要があります。パスは次のようになります：
https://{bucket}.s3.amazonaws.com/{prefix}/AWSDynamoDB/{export-id}/data/*
  • フォーマット：JSONEachRow
  • テーブル：スナップショットテーブル（例：上記の default.snapshot）

作成が完了すると、スナップショットテーブルと宛先テーブルにデータが含まれ始めます。次のステップに進む前にスナップショットのロードが完了するまで待つ必要はありません。

4. Kinesis ClickPipeを作成する​

次に、Kinesisストリームからのリアルタイムの変更をキャプチャするためのKinesis ClickPipeを設定できます。Kinesis ClickPipeガイドはこちらに従ってくださいが、次の設定を使用してください：

  • ストリーム：ステップ1で使用されたKinesisストリーム
  • テーブル：宛先テーブル（例：上記の default.destination）
  • オブジェクトのフラット化：true
  • カラムマッピング：
    • ApproximateCreationDateTime: version
    • その他のフィールドを適切な宛先カラムにマッピングする（下記のように）
DynamoDB Map Columns

5. クリーンアップ（オプション）​

スナップショットClickPipeが完了したら、スナップショットテーブルとマテリアライズドビューを削除できます。

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