Kinesis 커넥터

Kinesis 커넥터 (Amazon Kinesis Data Streams SQL Connector)

Kinesis 커넥터는 Amazon Kinesis Data Streams (KDS)에서 데이터를 읽고 쓸 수 있게 합니다. 스캔 소스는 무한(unbounded), 싱크는 배치/스트리밍 Append Mode로 동작합니다.

출처: 문서

본문

Kinesis 커넥터는 Amazon Kinesis Data Streams (KDS)에서 데이터를 읽고 쓸 수 있게 합니다.

의존성 (Dependencies)

Flink 버전 2.3용 커넥터는 아직 없습니다.

Kinesis 커넥터는 바이너리 배포판의 일부가 아닙니다. 클러스터 실행을 위해 연결하는 방법은 여기에서 확인하세요.

버저닝 (Versioning)

Kinesis 커넥터에는 두 가지 Table API 및 SQL 배포판이 있습니다. 이는 더 이상 사용되지 않는 SourceFunction 및 SinkFunction 인터페이스에서 새 Source 및 Sink 인터페이스로의 진행 중인 마이그레이션에서 비롯되었습니다.

Flink의 Table API 및 SQL 인터페이스는 커넥터 식별자당 하나의 TableFactory만 허용합니다. 애플리케이션의 의존성에는 식별자 kinesis를 가진 TableFactory가 하나만 포함될 수 있습니다.

다음 표는 선택한 배포판에 따라 사용되는 기본 인터페이스를 명확히 합니다:

flink-sql-connector-aws-kinesis-streams 또는 flink-sql-connector-kinesis 중 하나의 아티팩트만 포함하세요. 둘 다 포함하면 TableFactory 이름이 충돌합니다.

이 문서는 5.x 이상 버전을 대상으로 합니다. 주요 구성 섹션은 kinesis 식별자를 대상으로 합니다. 레거시 구성은 Configuration (kinesis-legacy) 문서를 참고하세요.

4.x와 5.x 사이에는 Table API와 SQL API 간 상태 호환성이 없습니다. 이는 기본 구현이 변경되었기 때문입니다.

v4.x kinesis 테이블로 작업을 중지한 시간보다 약간 이른 AT_TIMESTAMP의 source.init.position으로 v5.x kinesis 테이블로 작업을 시작하는 것을 고려하세요. 이는 일부 레코드가 재처리될 수도 있음을 유의하세요.

Kinesis 데이터 스트림 테이블 만들기 (How to create a Kinesis data stream table)

Amazon KDS Developer Guide의 지침을 따라 Kinesis 스트림을 설정하세요. 다음 예시는 Kinesis 데이터 스트림으로 백업되는 테이블을 만드는 방법을 보여줍니다:

CREATE TABLE KinesisTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP(3)
)
PARTITIONED BY (user_id, item_id)
WITH (
  'connector' = 'kinesis',
  'stream.arn' = 'arn:aws:kinesis:us-east-1:012345678901:stream/my-stream-name',
  'aws.region' = 'us-east-1',
  'source.init.position' = 'LATEST',
  'format' = 'csv'
);

사용 가능한 메타데이터 (Available Metadata)

kinesis 테이블 Source에는 VIRTUAL 컬럼이 지원되지 않는 알려진 버그가 있습니다. 수정이 완료될 때까지 kinesis-legacy를 사용하세요.

다음 메타데이터는 테이블 정의에서 읽기 전용(VIRTUAL) 컬럼으로 노출될 수 있습니다. 이는 kinesis-legacy 커넥터에서만 사용할 수 있습니다.

확장된 CREATE TABLE 예시는 이러한 메타데이터 필드를 노출하는 문법을 보여줍니다:

CREATE TABLE KinesisTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP(3),
  `arrival_time` TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL,
  `shard_id` VARCHAR(128) NOT NULL METADATA FROM 'shard-id' VIRTUAL,
  `sequence_number` VARCHAR(128) NOT NULL METADATA FROM 'sequence-number' VIRTUAL
)
PARTITIONED BY (user_id, item_id)
WITH (
  'connector' = 'kinesis-legacy',
  'stream' = 'user_behavior',
  'aws.region' = 'us-east-2',
  'scan.stream.initpos' = 'LATEST',
  'format' = 'csv'
);

커넥터 옵션 (Connector Options)

자세한 옵션(소스/싱크 시작 위치, 메타데이터, Data Type Mapping 등)은 공식 문서를 참고하세요. 레거시 kinesis-legacy 커넥터 옵션(scan.*, sink.* 레거시 옵션, shard.consumer.error.recoverable[0].exception, scan.watermark.* 등)도 동일 페이지에 문서화되어 있습니다.

레거시 커넥터 소스 옵션의 예:

싱크 옵션(레거시)의 예:

더 알아보기 (Learn more)

• Table 커넥터 개요
• Flink SQL 개요
• Amazon KDS Developer Guide