DLQ 및 메타데이터가 포함된 Apache Kafka¶

주요 기능¶

DLQ 및 메타데이터 커넥터가 포함된 Apache Kafka는 포괄적인 기능을 제공합니다.

• 데드 레터 큐(DLQ) 메시지 처리 실패 지원

• RECORD_METADATA 열에 Kafka 메시지 메타데이터를 추가합니다

• 구성 가능한 스키마화 - 스키마 탐지 활성화 또는 비활성화

• 스키마 진화를 통한 Iceberg 테이블 지원

• 여러 메시지 형식 - JSON 및 AVRO 지원

• AVRO 메시지에 대한 Schema Registry 통합

• 고급 패턴을 사용한 항목-테이블 매핑

• SASL 인증 지원

특정 매개 변수¶

Kafka용 Openflow Connector 설정하기 에 설명된 일반적인 매개 변수 외에도 이 커넥터에는 고급 기능을 위한 추가 매개 변수 컨텍스트가 포함되어 있습니다.

도식화 동작¶

커넥터의 동작은 Schematization Enabled 매개 변수에 따라 변경됩니다.

스키마 탐지 및 진화¶

커넥터는 스키마 감지 및 진화를 지원합니다. 커넥터가 로딩하는 새로운 데이터의 구조를 지원하도록 Snowflake의 테이블 구조를 자동으로 정의하고 발전시킬 수 있습니다.

스키마 감지 및 진화 기능이 없으면 커넥터가 로딩한 Snowflake 테이블은 RECORD_CONTENT 및 RECORD_METADATA 의 2개 OBJECT 열로만 구성됩니다.

스키마 감지 및 진화가 활성화되면 Snowflake가 스트리밍 데이터의 스키마를 감지하고 사용자 정의 스키마와 자동으로 일치하는 테이블에 데이터를 로드할 수 있습니다. 또한 Snowflake를 사용해 새로운 열을 추가하거나 새 데이터 파일에 누락된 열에서 NOT NULL 제약 조건을 삭제할 수도 있습니다.

커넥터를 통한 스키마 감지는 제공된 Schema Registry의 유무에 관계없이 지원됩니다. Schema Registry(Avro)를 사용하는 경우 제공된 Schema Registry에 정의된 데이터 타입으로 열이 생성됩니다. 스키마 레지스트리(JSON)가 없으면 제공된 데이터를 기반으로 데이터 타입이 유추됩니다.

추가적인 스키마화에는 JSON ARRAY가 지원되지 않습니다.

데드 레터 큐(DLQ)¶

DLQ 기능은 성공적으로 처리할 수 없는 메시지를 처리합니다.

Iceberg 테이블 지원¶

Iceberg 활성화됨 이 true 로 설정된 경우, Kafka용 Openflow Connector는 Snowflake 관리 Apache Iceberg™ 테이블 로 데이터를 수집할 수 있습니다.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    record_metadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "date_added": "2024-10-15"
}

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    id INT,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    RECORD_CONTENT OBJECT(
        id INT,
        body_temperature FLOAT,
        name STRING,
        approved_coffee_types ARRAY(STRING),
        animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
        date_added DATE
    )
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

사용 사례¶

이 커넥터는 다음에 이상적입니다.

• 프로덕션 환경 에 DLQ 가 필요한 경우

• Kafka 메타데이터가 중요한 데이터 계보 및 감사 의 경우

• 스키마 진화 요구 사항이 있는 복잡한 메시지 처리 의 경우

• Iceberg 테이블 통합

메타데이터나 DLQ 기능 없이 더 간단한 수집이 필요한 경우, JSON/AVRO 데이터 형식을 위한 Apache Kafka 커넥터를 사용하는 것이 좋습니다.