CDC 스트림 시작하기

이 가이드에서는 Aurora DSQL 클러스터에서 Amazon Kinesis 데이터 스트림으로 커밋된 행 수준 변경 사항 스트리밍을 시작하는 데 필요한 모든 단계를 안내합니다. 이 가이드가 끝날 때까지 유효한 CDC 파이프라인과 변경 레코드를 읽고 인쇄하는 Python 스크립트를 생성했습니다.

사전 조건

시작하기 전에 다음을 확인합니다.

1단계: Amazon Kinesis 데이터 스트림 생성

Aurora DSQL 클러스터와 동일한 AWS 계정 및 리전에 Kinesis 데이터 스트림을 생성합니다. JSON 형식에는 열 이름, 메타데이터 및 인코딩 오버헤드가 포함되므로 CDC 레코드는 해당 Aurora DSQL 행보다 큽니다.

Kinesis 데이터 스트림 크기 조정

Aurora DSQL CDC는 모든 변경 사항에 대해 전체 행을 제공합니다. 단일 열을 터치하는 업데이트는 행의 모든 열을 포함하는 레코드를 생성합니다. 레코드 삭제는 예외이며 프라이머리 키 열만 포함합니다.

평균 레코드 크기 추정

평균 온디스크 행 크기를 측정하여 CDC가 생성할 볼륨을 이해하고 크기 초과 레코드를 예측합니다. 다음 쿼리는 테이블의 평균 튜플 크기를 바이트 단위로 반환합니다.

SELECT avg(pg_column_size(t.*)) FROM your_table t;

CDC 레코드 봉투는 행 크기 위에 열 이름, 메타데이터 및 인코딩 오버헤드를 추가합니다. 정확한 레코드 형식은 레코드 페이로드 섹션을 참조하세요. Aurora DSQL이 Kinesis 레코드 크기 제한을 초과하는 레코드를 처리하는 방법은 크기 초과 레코드 처리 섹션을 참조하세요. Kinesis 서비스 제한의 전체 세트는 Amazon Kinesis Data Streams 개발자 안내서의 Amazon Kinesis Data Streams 할당량 및 제한을 참조하세요.

AWS/Kinesis 네임스페이스에서 IncomingBytes 및 WriteProvisionedThroughputExceeded에 CloudWatch 경보를 생성합니다. Kinesis 스로틀링은 CDC 전송 속도를 늦추고 복제 지연을 증가시킵니다. Aurora DSQL 측 지표 및 경보 지침은 모니터링 모범 사례 섹션을 참조하세요.

다음 예제에서는 AWS CLI를 사용합니다. AWS CLI 버전이 --max-record-size-in-ki-b 파라미터를 지원하지 않는 경우 AWS SDK를 사용하여 Kinesis CreateStream 작업을 직접적으로 호출합니다.

aws kinesis create-stream \
  --stream-name my-cdc-stream \
  --stream-mode-details StreamMode=ON_DEMAND \
  --max-record-size-in-ki-b 10240 \
  --region region

스트림이 다시 활성 상태가 될 때까지 대기:

aws kinesis describe-stream-summary \
  --stream-name my-cdc-stream \
  --region region \
  --query 'StreamDescriptionSummary.StreamStatus'

명령은 스트림이 준비되면 "ACTIVE"를 반환합니다.

출력에서 스트림 ARN을 기록합니다. 다음 단계에서 필요합니다. ARN의 형식은 arn:aws:kinesis:region:account-id:stream/my-cdc-stream입니다.

2단계: Aurora DSQL에 대한 IAM 역할 생성

Aurora DSQL은 IAM 역할을 수임하여 Kinesis 데이터 스트림에 CDC 레코드를 작성합니다. 이 단계에서는 신뢰 정책을 사용하여 역할을 생성하고 권한 정책을 연결합니다. 각 정책 요소에 대한 전체 설명은 IAM 구성 섹션을 참조하세요.

신뢰 정책 파일 생성

다음 JSON을 trust-policy.json으로 저장합니다. your-account-id, region 및 cluster-id를 해당 값으로 바꿉니다.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DSQLAccess",
            "Effect": "Allow",
            "Principal": {
                "Service": "dsql.amazonaws.com"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringEquals": {
                    "aws:SourceAccount": "your-account-id"
                },
                "ArnLike": {
                    "aws:SourceArn": "arn:aws:dsql:region:your-account-id:cluster/cluster-id/stream/*"
                }
            }
        }
    ]
}

역할 만들기

다음 명령을 실행하여 IAM 역할을 만듭니다.

aws iam create-role \
  --role-name dsql-cdc-role \
  --assume-role-policy-document file://trust-policy.json

권한 정책 파일 생성

다음 JSON을 permissions-policy.json으로 저장합니다. 자리 표시자 값을 Kinesis 데이터 스트림 ARN으로 바꿉니다. 이 KMSAccess 문은 Kinesis 데이터 스트림이 AWS KMS 고객 관리형 키를 사용하는 경우에만 필요하지만, 나중에 고객 관리형 키를 추가해도 CDC 스트림이 손상되지 않도록 선제적으로 포함할 수 있습니다. 각 조건에 관한 완전한 설명을 보려면 서비스 역할 권한 정책 섹션을 참조하세요.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "KinesisAccess",
            "Effect": "Allow",
            "Action": [
                "kinesis:PutRecord",
                "kinesis:PutRecords",
                "kinesis:DescribeStreamSummary",
                "kinesis:ListShards"
            ],
            "Resource": "arn:aws:kinesis:region:your-account-id:stream/my-cdc-stream"
        },
        {
            "Sid": "KMSAccess",
            "Effect": "Allow",
            "Action": [
                "kms:GenerateDataKey"
            ],
            "Resource": "arn:aws:kms:*:*:key/*",
            "Condition": {
                "StringEquals": {
                    "kms:ViaService": "kinesis.region.amazonaws.com",
                    "kms:EncryptionContext:aws:kinesis:arn": "arn:aws:kinesis:region:your-account-id:stream/my-cdc-stream",
                    "aws:ResourceAccount": "${aws:PrincipalAccount}"
                }
            }
        }
    ]
}

권한 정책 연결

다음 명령을 실행합니다.

aws iam put-role-policy \
  --role-name dsql-cdc-role \
  --policy-name dsql-cdc-kinesis-access \
  --policy-document file://permissions-policy.json

create-role 출력에서 역할 ARN을 기록합니다. ARN의 형식은 arn:aws:iam::your-account-id:role/dsql-cdc-role입니다.

3단계: CDC 스트림 생성

AWS CLI를 사용하여 Aurora DSQL 클러스터를 Kinesis 데이터 스트림에 연결하는 CDC 스트림을 생성합니다. 자리 표시자 값을 1단계의 Kinesis 스트림 ARN, 2단계의 IAM 역할 ARN 및 클러스터 식별자로 바꿉니다.

aws dsql create-stream \
  --cluster-identifier cluster-id \
  --target-definition '{"kinesis":{"streamArn":"kinesis-stream-arn","roleArn":"role-arn"}}' \
  --ordering UNORDERED \
  --format JSON \
  --tags '{"Name":"my-cdc-stream"}' \
  --region region

응답에는 스트림 식별자와 CREATING 상태가 포함됩니다. 스트림 생성에는 일반적으로 1~3분이 소요됩니다.

스트림이 다시 활성 상태가 될 때까지 대기

스트림 상태가 ACTIVE에 도달할 때까지 폴링합니다.

aws dsql get-stream \
  --cluster-identifier cluster-id \
  --stream-identifier stream-id \
  --region region \
  --query 'status'

AWS SDK의 StreamActive 웨이터를 사용하여 자동으로 폴링할 수도 있습니다.

스트림이 ACTIVE에 도달하면 Aurora DSQL은 커밋된 행 수준 변경 사항을 Kinesis 데이터 스트림에 전송하기 시작합니다.

4단계: 레코드가 흐르고 있는지 확인

Aurora DSQL 클러스터의 테이블에 행을 삽입합니다. 예제:

CREATE TABLE IF NOT EXISTS test_cdc (
    id INT PRIMARY KEY,
    message TEXT
);

INSERT INTO test_cdc VALUES (1, 'hello cdc');

Kinesis 데이터 스트림에서 읽어 CDC 레코드가 도착했는지 확인합니다.

SHARD_ITERATOR=$(aws kinesis get-shard-iterator \
  --stream-name my-cdc-stream \
  --shard-id shardId-000000000000 \
  --shard-iterator-type TRIM_HORIZON \
  --region region \
  --query 'ShardIterator' --output text)

aws kinesis get-records \
  --shard-iterator "$SHARD_ITERATOR" \
  --region region

각 레코드의 Data 필드에는 JSON 페이로드가 포함됩니다. AWS CLI를 사용하면 응답에서 페이로드가 Base64로 인코딩됩니다. boto3 SDK를 사용하면 SDK가 자동으로 디코딩합니다. 디코딩된 JSON은 다음과 같습니다.

{
    "type": "full",
    "op": "c",
    "before": null,
    "after": {"id": 1, "message": "hello cdc"},
    "source": {
        "version": "1.0",
        "ts_ms": 1705318200000,
        "ts_ns": 1705318200000000000,
        "txId": "ffthunp5stx6ffs2vyfqoatmfu",
        "schema": "public",
        "table": "test_cdc",
        "db": "postgres",
        "cluster": "cluster-id"
    },
    "ts_ms": 1705318200125,
    "ts_ns": 1705318200125483291
}

각 필드에 대한 완전한 설명은 CDC 레코드 이해 섹션을 참조하세요.

5단계: Python 스크립트로 레코드 사용

다음 Python 스크립트는 Kinesis 데이터 스트림에서 CDC 레코드를 읽고 각 변경 이벤트를 인쇄합니다. 스크립트는 boto3 Amazon Kinesis 클라이언트를 사용하여 샤드를 반복하고 각 레코드를 디코딩합니다. Aurora DSQL CDC는 최소 1번 전송을 사용하므로 스크립트는 동일한 레코드를 두 번 이상 인쇄할 수 있습니다.

"""
Read CDC records from an Amazon Kinesis data stream.

Usage:
    pip install boto3
    python consume_cdc.py --stream-name my-cdc-stream --region us-east-1
"""
from __future__ import annotations

import argparse
import json

import boto3


def consume_cdc(stream_name: str, region: str) -> None:
    kinesis = boto3.client("kinesis", region_name=region)

    # List all shards (paginate if the stream has many shards)
    shard_ids: list[str] = []
    paginator = kinesis.get_paginator("list_shards")
    for page in paginator.paginate(StreamName=stream_name):
        shard_ids.extend(s["ShardId"] for s in page["Shards"])
    print(f"Reading from {stream_name} ({len(shard_ids)} shard(s))")

    for shard_id in shard_ids:
        iterator_response = kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType="TRIM_HORIZON",
        )
        shard_iterator = iterator_response["ShardIterator"]

        while shard_iterator:
            records_response = kinesis.get_records(
                ShardIterator=shard_iterator, Limit=100
            )
            shard_iterator = records_response.get("NextShardIterator")

            for record in records_response["Records"]:
                # boto3 decodes Base64 automatically; record["Data"] is bytes.
                payload = json.loads(record["Data"])

                # A record's "type" field identifies its structure.
                # "full": inlined record with before/after values.
                # "chunked": main record that references fragments for a split image.
                # "fragment": one piece of a chunked image; reassemble in production code.
                # For details, see cdc-record-format.html#cdc-oversized-records.
                record_type = payload.get("type", "full")
                if record_type == "fragment":
                    print(f"[FRAGMENT] chunk_id={payload['chunk_id']} index={payload['index']}")
                    continue

                source = payload["source"]
                op = payload["op"]
                ts_ns = source["ts_ns"]
                tx_id = source["txId"]
                table = f"{source['schema']}.{source['table']}"

                # Aurora DSQL currently emits "c" for both inserts and updates. A subsequent
                # release will emit "u" for updates, and "c" for inserts. Design your
                # consumer to handle all three values; this map stays correct across the
                # transition.
                op_labels = {"c": "INSERT/UPDATE", "u": "UPDATE", "d": "DELETE"}
                print(
                    f"[{op_labels.get(op, op)}] {table} "
                    f"txId={tx_id} ts_ns={ts_ns} type={record_type}"
                )
                if payload.get("after"):
                    print(f"  after:  {json.dumps(payload['after'])}")
                if payload.get("before"):
                    print(f"  before: {json.dumps(payload['before'])}")
                if record_type == "chunked":
                    print(f"  chunked: {json.dumps(payload['chunked'])}")

            if not records_response["Records"]:
                break  # No more records in this shard


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Consume DSQL CDC records from Kinesis"
    )
    parser.add_argument("--stream-name", required=True, help="Kinesis stream name")
    parser.add_argument("--region", required=True, help="AWS Region")
    args = parser.parse_args()
    consume_cdc(args.stream_name, args.region)

스크립트를 실행합니다.

pip install boto3
python consume_cdc.py \
  --stream-name my-cdc-stream \
  --region region

스크립트는 각 변경 이벤트가 도착하면 인쇄합니다. 출력은 다음과 비슷합니다.

Reading from my-cdc-stream (4 shard(s))
[INSERT/UPDATE] public.test_cdc txId=ffthunp5stx6ffs2vyfqoatmfu ts_ns=1705318200000000000 type=full
  after:  {"id": 1, "message": "hello cdc"}

마지막 쓰기 우선 중복 제거 추가

Aurora DSQL CDC는 최소 1번 전송을 사용하므로 프로덕션 앱은 레코드를 중복 제거 및 순서 지정해야 합니다. 다음 코드 예제는 최고 수위선 접근 방식을 보여줍니다. 각 프라이머리 키에 대해 지금까지 관찰된 가장 높은 source.ts_ns를 추적하고 타임스탬프가 같거나 이전인 모든 레코드를 삭제합니다. PK_COLUMNS를 처리 중인 테이블의 프라이머리 키 열 이름으로 설정합니다. 여러 테이블 또는 삭제를 처리하는 전략은 소비자 전략 섹션을 참조하세요.

# Set PK_COLUMNS to the primary key column(s) of your table.
PK_COLUMNS = ["id"]

# Maps each primary key value to the highest ts_ns seen for that key.
high_water: dict[tuple, int] = {}

def process_record(payload: dict) -> bool:
    """Return True if the record is new, False if it's a duplicate or stale.

    Skip fragment records; reassemble them into a full image before calling this.
    """
    if payload.get("type") == "fragment":
        return False  # Fragments are reassembled upstream, not deduplicated here.

    source = payload["source"]
    ts_ns = source["ts_ns"]
    op = payload["op"]

    # For inserts/updates the row is in "after"; for deletes it's in "before".
    row = payload.get("after") or payload.get("before") or {}
    pk = tuple(row.get(col) for col in PK_COLUMNS)

    prev_ts = high_water.get(pk, -1)
    if ts_ns <= prev_ts:
        return False  # Duplicate or out-of-order record

    high_water[pk] = ts_ns
    return True

CDC 스트림 관리

스트림 나열

클러스터의 모든 CDC 스트림을 나열하려면 ListStreams 작업을 사용합니다.

aws dsql list-streams \
  --cluster-identifier cluster-id \
  --region region

스트림 삭제

CDC 스트림을 삭제하려면 다음 명령을 실행합니다.

aws dsql delete-stream \
  --cluster-identifier cluster-id \
  --stream-identifier stream-id \
  --region region

Aurora DSQL이 스트림을 완전히 삭제했음을 나타내는 ResourceNotFoundException가 반환될 때까지 StreamNotExists 웨이터를 사용하여 GetStream을 폴링할 수 있습니다.