DynamoDB에서 ClickHouse로 CDC

이 페이지에서는 ClickPipes를 사용하여 DynamoDB에서 ClickHouse로 CDC를 설정하는 방법을 설명합니다. 이 통합은 2개의 구성 요소로 이루어집니다:

S3 ClickPipes를 통한 초기 스냅샷
Kinesis ClickPipes를 통한 실시간 업데이트

데이터는 ReplacingMergeTree에 수집됩니다. 이 테이블 엔진은 업데이트 작업을 적용할 수 있어 CDC 시나리오에서 일반적으로 사용됩니다. 이 패턴에 대한 자세한 내용은 다음 블로그 글에서 확인할 수 있습니다:

Kinesis 스트림 설정

먼저 DynamoDB 테이블에서 Kinesis 스트림을 활성화하여 변경 사항을 실시간으로 캡처해야 합니다. 데이터가 누락되지 않도록 스냅샷을 생성하기 전에 이 작업을 수행하는 것이 좋습니다. AWS 가이드는 여기에서 확인하십시오.

스냅샷 생성

다음으로 DynamoDB 테이블의 스냅샷을 생성합니다. 이는 AWS에서 S3로 내보내기를 통해 수행할 수 있습니다. AWS 가이드는 여기에서 확인할 수 있습니다. DynamoDB JSON 포맷으로 “Full export”를 수행해야 합니다.

ClickHouse에 스냅샷 로드하기

필요한 테이블 생성

DynamoDB의 스냅샷 데이터는 다음과 같은 형식입니다:

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

데이터가 중첩된 형식이라는 점에 유의하십시오. 이 데이터를 ClickHouse에 로드하기 전에 평탄화해야 합니다. 이는 ClickHouse의 materialized view에서 JSONExtract 함수를 사용해 수행할 수 있습니다.다음과 같이 3개의 테이블을 생성해야 합니다:

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 생성

이제 S3의 스냅샷 데이터를 ClickHouse로 로드하는 ClickPipe를 생성할 수 있습니다. 여기의 S3 ClickPipe 가이드를 따르되, 다음 설정을 사용하십시오:

수집 경로: S3에서 내보낸 JSON 파일의 경로를 찾아야 합니다. 경로는 다음과 같은 형식입니다:

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

포맷: JSONEachRow
테이블: 스냅샷 테이블(예: 위 예시의 default.snapshot)

생성이 완료되면 스냅샷 테이블과 대상 테이블에 데이터가 적재되기 시작합니다. 다음 단계로 진행하기 전에 스냅샷 로드가 완료될 때까지 기다릴 필요는 없습니다.

Kinesis ClickPipe 생성

이제 Kinesis 스트림의 실시간 변경 사항을 캡처하도록 Kinesis ClickPipe를 설정할 수 있습니다. 여기의 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";

ClickPipes for BigQuery

프로그래밍 방식 액세스

기타

필요한 테이블 생성

스냅샷 ClickPipe 생성

​필요한 테이블 생성

​스냅샷 ClickPipe 생성

필요한 테이블 생성

스냅샷 ClickPipe 생성