从 DynamoDB 到 ClickHouse 的 CDC（变更数据捕获）

本页介绍如何使用 ClickPipes 配置从 DynamoDB 到 ClickHouse 的 CDC (变更数据捕获) (变更数据捕获) 。此集成包含 2 个组件：

通过 S3 ClickPipes 执行初始快照
通过 Kinesis ClickPipes 进行实时更新

数据将被摄取到 ReplacingMergeTree 中。该表引擎常用于 CDC (变更数据捕获) 场景，以便应用更新操作。有关此模式的更多信息，请参阅以下博客文章：

设置 Kinesis 数据流

首先，您需要在 DynamoDB 表上启用 Kinesis 数据流，以实时捕获变更。我们希望在创建快照之前先完成这一步，以免遗漏任何数据。 AWS 指南可见此处。

创建快照

接下来，我们将为 DynamoDB 表创建一个快照。这可以通过 AWS 导出到 S3 的方式来实现。可在此处查看 AWS 指南。 你需要以 DynamoDB JSON 格式执行“完整导出”。

将快照导入 ClickHouse

创建所需的表

来自 DynamoDB 的快照数据大致如下：

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

请注意，这些数据采用嵌套格式。在加载到 ClickHouse 之前，需要先将其展平。可以通过在 ClickHouse 的 materialized view 中使用 JSONExtract 函数来实现。我们需要创建三个表：

一个用于存储来自 DynamoDB 的原始数据的表
一个用于存储最终展平后数据的表 (目标表)
一个用于展平数据的 materialized view

对于上面的 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

现在，您可以创建一个 ClickPipe，将快照数据从 S3 加载到 ClickHouse。请遵循此处的 S3 ClickPipe 指南，但使用以下设置：

摄取路径：您需要找到 S3 中已导出的 JSON 文件路径。该路径大致如下：

https://{bucket}.s3.amazonaws.com/{prefix}/AWSDynamoDB/{export-id}/data/*

格式: JSONEachRow
表: 你的快照表 (例如上面示例中的 default.snapshot)

创建完成后，数据会开始写入快照表和目标端表。你无需等到快照加载完成后再继续下一步。

创建 Kinesis ClickPipe

现在我们可以设置 Kinesis ClickPipe，以捕获来自 Kinesis 数据流的实时变更。请按照这里的 Kinesis ClickPipe 指南操作，但使用以下设置：

Stream：第 1 步中使用的 Kinesis 数据流
Table：你的目标端表 (例如上面示例中的 default.destination)
Flatten object：true
Column mappings：
- ApproximateCreationDateTime：version
- 按照下图所示，将其他字段映射到相应的目标端列

清理（可选）

快照 ClickPipe 完成后，您可以删除快照表和 materialized view。

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

最后修改于 2026年7月24日

监控使用 Prometheus 指标监控 ClickPipes。

ClickPipes for BigQuery

程序化访问

其他

从 DynamoDB 到 ClickHouse 的 CDC（变更数据捕获）

创建所需的表

创建快照 ClickPipe

​创建所需的表

​创建快照 ClickPipe

创建所需的表

创建快照 ClickPipe