> ## 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.

> O ClickPipes permite conectar o ClickHouse ao DynamoDB.

# CDC do DynamoDB para o ClickHouse

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>
            Sem suporte no ClickHouse Cloud
        </div>;
};

Esta página explica como configurar CDC do DynamoDB para o ClickHouse usando ClickPipes. Há 2 componentes nesta integração:

1. O snapshot inicial via S3 ClickPipes
2. Atualizações em tempo real via Kinesis ClickPipes

Os dados serão inseridos em uma `ReplacingMergeTree`. Esse mecanismo de tabela é comumente usado em cenários de CDC para permitir a aplicação de operações de atualização. Saiba mais sobre esse padrão nos seguintes artigos do blog:

* [Captura de Dados de Alteração (CDC) com PostgreSQL e ClickHouse - Parte 1](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-1?loc=docs-rockest-migrations)
* [Captura de Dados de Alteração (CDC) com PostgreSQL e ClickHouse - Parte 2](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2?loc=docs-rockest-migrations)

<Steps>
  <Step title="Configure o Kinesis stream" id="1-set-up-kinesis-stream">
    Primeiro, habilite um Kinesis Stream na sua tabela do DynamoDB para capturar alterações em tempo real. É importante fazer isso antes de criar o snapshot para evitar a perda de dados.
    Consulte o guia da AWS [aqui](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="Kinesis Stream do DynamoDB" border width="1238" height="215" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-kinesis-stream.webp" />
  </Step>

  <Step title="Crie o snapshot" id="2-create-the-snapshot">
    Em seguida, criaremos um snapshot da tabela do DynamoDB. Isso pode ser feito por meio de uma exportação do AWS para o S3. Encontre o guia da AWS [aqui](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.HowItWorks.html).
    **Você vai querer fazer uma "Exportação completa" no formato DynamoDB JSON.**

    <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="Exportação do DynamoDB para S3" border width="686" height="929" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-s3-export.webp" />
  </Step>

  <Step title="Carregue o snapshot no ClickHouse" id="3-load-the-snapshot-into-clickhouse">
    ### Crie as tabelas necessárias

    Os dados do snapshot do DynamoDB terão esta aparência:

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

    Observe que os dados estão em um formato aninhado. Precisaremos desaninhar esses dados antes de carregá-los no ClickHouse. Isso pode ser feito usando a função `JSONExtract` no ClickHouse em uma visão materializada.

    Vamos criar três tabelas:

    1. Uma tabela para armazenar os dados brutos do DynamoDB
    2. Uma tabela para armazenar os dados finais desaninhados (tabela de destino)
    3. Uma visão materializada para desaninhar os dados

    Para os dados de exemplo do DynamoDB acima, as tabelas do ClickHouse ficariam assim:

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

    Há alguns requisitos para a tabela de destino:

    * Essa tabela deve ser uma tabela `ReplacingMergeTree`
    * A tabela deve ter uma coluna `version`
      * Nas etapas seguintes, faremos o mapeamento do campo `ApproximateCreationDateTime` do Kinesis stream para a coluna `version`.
    * A tabela deve usar a chave de partição como chave de ordenação (especificada por `ORDER BY`)
      * Linhas com a mesma chave de ordenação serão deduplicadas com base na coluna `version`.

    ### Crie o ClickPipe de snapshot

    Agora você pode criar um ClickPipe para carregar os dados de snapshot do S3 no ClickHouse. Siga o guia do S3 ClickPipe [aqui](/docs/pt-BR/integrations/clickpipes/object-storage/amazon-s3/overview), mas use as seguintes configurações:

    * **Caminho de ingestão**: Você precisará localizar o caminho dos arquivos JSON exportados no S3. O caminho será parecido com isto:

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

    * **Formato**: JSONEachRow
    * **Tabela**: Sua tabela de snapshot (por exemplo, `default.snapshot` no exemplo acima)

    Depois de criados, os dados começarão a ser preenchidos nas tabelas de snapshot e de destino. Você não precisa esperar o carregamento do snapshot terminar antes de prosseguir para a próxima etapa.
  </Step>

  <Step title="Crie o ClickPipe para o Kinesis" id="4-create-the-kinesis-clickpipe">
    Agora podemos configurar o ClickPipe do Kinesis para capturar alterações em tempo real do Kinesis stream. Siga o guia do ClickPipe do Kinesis [aqui](/docs/pt-BR/integrations/clickpipes/kinesis/overview), mas use as seguintes configurações:

    * **Stream**: O Kinesis stream usado na etapa 1
    * **Table**: Sua tabela de destino (por exemplo, `default.destination` no exemplo acima)
    * **Flatten object**: true
    * **Column mappings**:
      * `ApproximateCreationDateTime`: `version`
      * Mapeie os outros campos para as colunas de destino apropriadas, conforme mostrado abaixo

    <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="Mapeamento de colunas do DynamoDB" border width="1784" height="1630" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-map-columns.webp" />
  </Step>

  <Step title="Limpeza (opcional)" id="5-cleanup-optional">
    Depois que o ClickPipe de snapshot for concluído, você poderá excluir a tabela de snapshot e a visão materializada.

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