> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

> ClickPipes を使用すると、DynamoDB を ClickHouse に接続できます。

# DynamoDB から ClickHouse への CDC

export const Image = ({img, alt, size = "lg"}) => {
  const normalizedSize = ["sm", "md", "lg"].includes(size) ? size : "lg";
  return <div className={`ch-image-${normalizedSize}`}>
      <Frame>
        <img src={img} alt={alt} />
      </Frame>
    </div>;
};

export const CloudNotSupportedBadge = () => {
  return <div className="cloudNotSupportedBadge">
            <div className="cloudNotSupportedIcon">
            <svg width="16" height="16" viewBox="0 0 16 16" fill="none" xmlns="http://www.w3.org/2000/svg">
                <path strokeWidth="1.5" d="M6.33366 12.6666L12.3739 12.6667C13.6593 12.6667 14.7073 11.6187 14.7073 10.3334C14.7073 9.04804 13.6593 8.00003 12.3739 8.00003C12.3739 8.00003 12.3337 7.66659 12.0003 7.33325M10.667 5.33322C8.00033 2.33325 4.45395 4.78537 4.14195 6.68203C2.55728 6.7627 1.29395 8.06203 1.29395 9.6667C1.29395 11.3234 2.66699 12.6666 4.00033 12.6666" stroke="currentColor" strokeLinecap="round" strokeLinejoin="round" />
                <path strokeWidth="1.5" d="M2.66699 14L12.0003 4.66663" stroke="currentColor" strokeLinecap="round" strokeLinejoin="round" />
            </svg>

        </div>
            ClickHouse Cloud では利用できません
        </div>;
};

このページでは、ClickPipes を使用して DynamoDB から ClickHouse への CDC (変更データキャプチャ) を設定する方法を説明します。このインテグレーションは、次の 2 つのコンポーネントで構成されています。

1. S3 ClickPipes による初期スナップショット
2. Kinesis ClickPipes によるリアルタイム更新

データは `ReplacingMergeTree` に取り込まれます。このテーブルエンジンは、更新操作を適用できるようにするため、CDC のユースケースで一般的に使用されます。このパターンの詳細については、以下のブログ記事を参照してください。

* [PostgreSQL と ClickHouse を使った変更データキャプチャ (CDC) - 第1部](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-1?loc=docs-rockest-migrations)
* [PostgreSQL と ClickHouse を使った変更データキャプチャ (CDC) - 第2部](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2?loc=docs-rockest-migrations)

<Steps>
  <Step title="Kinesis Stream を設定する" id="1-set-up-kinesis-stream">
    まず、変更をリアルタイムで取得するために、DynamoDB テーブルで Kinesis Stream を有効にします。データの取りこぼしを防ぐため、スナップショットを作成する前にこれを行います。
    AWS のガイドは[こちら](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/kds.html)にあります。

    <Image img="https://mintcdn.com/private-7c7dfe99/dZZG_-B0EzCG8L4V/images/integrations/data-ingestion/dbms/dynamodb/dynamodb-kinesis-stream.webp?fit=max&auto=format&n=dZZG_-B0EzCG8L4V&q=85&s=939ff3cb23f96ab215db4148fa61f351" size="lg" alt="DynamoDB Kinesis Stream" border width="1238" height="215" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-kinesis-stream.webp" />
  </Step>

  <Step title="スナップショットを作成する" id="2-create-the-snapshot">
    次に、DynamoDB テーブルのスナップショットを作成します。これは、AWS から S3 にエクスポートすることで実行できます。AWS のガイドは[こちら](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.HowItWorks.html)にあります。
    **DynamoDB JSON フォーマットで「Full export」を実行してください。**

    <Image img="https://mintcdn.com/private-7c7dfe99/dZZG_-B0EzCG8L4V/images/integrations/data-ingestion/dbms/dynamodb/dynamodb-s3-export.webp?fit=max&auto=format&n=dZZG_-B0EzCG8L4V&q=85&s=483c56ff98809c4b31281691b80b47ae" size="md" alt="DynamoDB S3 Export" border width="686" height="929" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-s3-export.webp" />
  </Step>

  <Step title="スナップショットを ClickHouse に読み込む" id="3-load-the-snapshot-into-clickhouse">
    ### 必要なテーブルを作成する

    DynamoDB から取得したスナップショットデータは、次のようになります:

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

    データがネストされた形式になっていることに注目してください。このデータを ClickHouse に読み込む前に、フラット化する必要があります。これは、ClickHouse の materialized view で `JSONExtract` 関数を使って行えます。

    作成するテーブルは 3 つあります。

    1. DynamoDB の生データを格納するためのテーブル
    2. 最終的なフラット化済みデータを格納するためのテーブル (宛先テーブル)
    3. データをフラット化するための materialized view

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

    ```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;
    ```

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

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

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

    これで、S3 から ClickHouse にスナップショットデータを読み込むための ClickPipe を作成できます。[こちら](/docs/ja/integrations/clickpipes/object-storage/amazon-s3/overview)の S3 ClickPipe ガイドに従ってください。ただし、以下の設定を使用します。

    * **取り込みパス**: S3 内でエクスポートされた JSON ファイルのパスを特定する必要があります。パスは次のようになります:

    ```text theme={null}
    https://{bucket}.s3.amazonaws.com/{prefix}/AWSDynamoDB/{export-id}/data/*
    ```

    * **フォーマット**: JSONEachRow
    * **テーブル**: スナップショットテーブル (上記の例では `default.snapshot`)

    作成が完了すると、スナップショットテーブルと宛先テーブルへのデータ投入が始まります。スナップショットの読み込みが完了するのを待たずに、次の手順に進んでかまいません。
  </Step>

  <Step title="Kinesis ClickPipeを作成する" id="4-create-the-kinesis-clickpipe">
    これで、Kinesis Stream からリアルタイムの変更を取り込むための Kinesis ClickPipe を設定できます。[こちら](/docs/ja/integrations/clickpipes/kinesis/overview)の Kinesis ClickPipe ガイドに従ってください。ただし、以下の設定を使用します。

    * **Stream**: 手順 1 で使用した Kinesis Stream
    * **Table**: 宛先テーブル (例: 上記の `default.destination`)
    * **Flatten object**: true
    * **Column mappings**:
      * `ApproximateCreationDateTime`: `version`
      * 以下に示すように、他のフィールドを適切な宛先カラムにマッピングします

    <Image img="https://mintcdn.com/private-7c7dfe99/dZZG_-B0EzCG8L4V/images/integrations/data-ingestion/dbms/dynamodb/dynamodb-map-columns.webp?fit=max&auto=format&n=dZZG_-B0EzCG8L4V&q=85&s=0261fe733d26d110dd5490b47d78a71b" size="md" alt="DynamoDB カラムのマッピング" border width="1784" height="1630" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-map-columns.webp" />
  </Step>

  <Step title="クリーンアップ（任意）" id="5-cleanup-optional">
    スナップショット用の ClickPipe が完了したら、スナップショットテーブルと materialized view を削除できます。

    ```sql theme={null}
    DROP TABLE IF EXISTS "default"."snapshot";
    DROP TABLE IF EXISTS "default"."snapshot_clickpipes_error";
    DROP VIEW IF EXISTS "default"."snapshot_mv";
    ```
  </Step>
</Steps>