> ## 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 permite conectar ClickHouse a DynamoDB.

# CDC de DynamoDB a 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>
            No es compatible con ClickHouse Cloud
        </div>;
};

Esta página explica cómo configurar CDC de DynamoDB a ClickHouse con ClickPipes. Esta integración consta de 2 componentes:

1. La instantánea inicial mediante S3 ClickPipes
2. Las actualizaciones en tiempo real mediante Kinesis ClickPipes

Los datos se ingestarán en un `ReplacingMergeTree`. Este motor de tabla se usa habitualmente en escenarios de CDC para permitir aplicar operaciones de actualización. Puede encontrar más información sobre este patrón en los siguientes artículos del blog:

* [Captura de datos de cambios (CDC) con PostgreSQL y ClickHouse - Parte 1](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-1?loc=docs-rockest-migrations)
* [Captura de datos de cambios (CDC) con PostgreSQL y ClickHouse - Parte 2](https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2?loc=docs-rockest-migrations)

<Steps>
  <Step title="Configurar el flujo de Kinesis" id="1-set-up-kinesis-stream">
    Primero, querrá habilitar un Kinesis Stream en su tabla de DynamoDB para capturar los cambios en tiempo real. Queremos hacerlo antes de crear la instantánea para evitar perder datos.
    Puede consultar la guía de AWS [aquí](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="Crear la instantánea" id="2-create-the-snapshot">
    A continuación, crearemos una instantánea de la tabla de DynamoDB. Esto puede hacerse mediante una exportación de AWS a S3. Consulte la guía de AWS disponible [aquí](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.HowItWorks.html).
    **Deberá realizar una "exportación completa" en el formato JSON de DynamoDB.**

    <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="Exportación de DynamoDB a S3" border width="686" height="929" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-s3-export.webp" />
  </Step>

  <Step title="Cargar la instantánea en ClickHouse" id="3-load-the-snapshot-into-clickhouse">
    ### Crear las tablas necesarias

    Los datos de la instantánea de DynamoDB tendrán este aspecto:

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

    Observe que los datos están en un formato anidado. Tendremos que aplanarlos antes de cargarlos en ClickHouse. Esto se puede hacer con la función `JSONExtract` de ClickHouse en una vista materializada.

    Querremos crear tres tablas:

    1. Una tabla para almacenar los datos sin procesar de DynamoDB
    2. Una tabla para almacenar los datos finales aplanados (tabla de destino)
    3. Una vista materializada para aplanar los datos

    Para los datos de ejemplo de DynamoDB mostrados arriba, las tablas de ClickHouse se verían así:

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

    Hay algunos requisitos para la tabla de destino:

    * Esta tabla debe ser de tipo `ReplacingMergeTree`
    * La tabla debe tener una columna `version`
      * En pasos posteriores, asignaremos el campo `ApproximateCreationDateTime` del flujo de Kinesis a la columna `version`.
    * La tabla debe usar la clave de partición como clave de ordenación (especificada por `ORDER BY`)
      * Las filas con la misma clave de ordenación se deduplicarán según la columna `version`.

    ### Crear el ClickPipe de instantánea

    Ahora puede crear un ClickPipe para cargar los datos de la instantánea desde S3 en ClickHouse. Siga la guía de S3 ClickPipe [aquí](/docs/es/integrations/clickpipes/object-storage/amazon-s3/overview), pero use la siguiente configuración:

    * **Ruta de ingesta**: Tendrá que localizar la ruta de los archivos JSON exportados en S3. La ruta tendrá un aspecto similar a este:

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

    * **Formato**: JSONEachRow
    * **Tabla**: Su tabla de instantánea (p. ej. `default.snapshot` en el ejemplo anterior)

    Una vez creadas, los datos comenzarán a cargarse en las tablas de instantánea y de destino. No es necesario esperar a que finalice la carga de la instantánea antes de pasar al siguiente paso.
  </Step>

  <Step title="Crear el ClickPipe para Kinesis" id="4-create-the-kinesis-clickpipe">
    Ahora podemos configurar el ClickPipe de Kinesis para capturar cambios en tiempo real del stream de Kinesis. Siga la guía de Kinesis ClickPipe [aquí](/docs/es/integrations/clickpipes/kinesis/overview), pero use la siguiente configuración:

    * **Stream**: El stream de Kinesis usado en el paso 1
    * **Table**: Su tabla de destino (por ejemplo, `default.destination` en el ejemplo anterior)
    * **Flatten object**: true
    * **Column mappings**:
      * `ApproximateCreationDateTime`: `version`
      * Asigne los demás campos a las columnas de destino correspondientes, como se muestra a continuación

    <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: asignación de columnas" border width="1784" height="1630" data-path="images/integrations/data-ingestion/dbms/dynamodb/dynamodb-map-columns.webp" />
  </Step>

  <Step title="Limpieza (opcional)" id="5-cleanup-optional">
    Una vez que haya finalizado el ClickPipe de instantánea, puede eliminar la tabla de instantánea y la vista 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>