Snowpipe Streaming Classic과 함께하는 Kafka 커넥터의 스키마 감지와 진화

Snowpipe Streaming Classic과 함께하는 Kafka 커넥터의 스키마 감지와 진화

Kafka 커넥터와 함께하는 Snowpipe Streaming에서 스키마 감지(schema detection)와 진화(evolution)가 어떻게 동작하는지 알려드릴게요. 이 기능을 켜면 Snowflake 테이블의 구조가 Kafka 커넥터가 로드하는 새 스트리밍 데이터 구조에 맞춰 자동으로 정의·진화할 수 있습니다.

출처: Snowflake 문서

본문

참고 Snowflake Connector for Kafka (v4)는 스키마 진화를 기본 지원합니다. 자세한 내용은 스키마 진화를 참고하세요.

Kafka 커넥터와 함께하는 Snowpipe Streaming은 스키마 감지와 진화를 지원합니다. Snowflake의 테이블 구조를 정의·진화시켜 Kafka 커넥터가 로드하는 새 Snowpipe 스트리밍 데이터의 구조를 자동으로 지원할 수 있어요.

스키마 감지·진화가 없으면 Kafka 커넥터가 로드하는 Snowflake 테이블은 RECORD_CONTENT와 RECORD_METADATA라는 두 VARIANT 컬럼만으로 구성됩니다. 스키마 감지·진화를 활성화하면 Snowflake가 스트리밍 데이터의 스키마를 감지해 사용자 정의 스키마와 자동으로 일치하는 테이블로 데이터를 로드할 수 있어요. Snowflake는 새 데이터 파일에 없는 컬럼에 새 컬럼을 추가하거나 NOT NULL 제약을 제거하는 것도 허용합니다.

참고 이 기능은 Kafka 커넥터와 함께하는 Snowpipe Streaming에서만 동작합니다. 파일 기반 Snowpipe를 사용하는 Kafka 커넥터는 지원하지 않아요.

사전 요구사항

이 기능을 활성화하기 전에 다음 사전 요구사항을 설정하세요.

필요한 Kafka 속성 구성

Kafka 커넥터 속성 파일에서 다음 필수 속성을 구성하세요.

snowflake.ingestion.method

Kafka 토픽 데이터를 로드하려면 SNOWPIPE_STREAMING을 지정하세요. 이 기능은 SNOWPIPE를 지원하지 않습니다.

snowflake.enable.schematization

Snowpipe Streaming과 함께하는 Kafka 커넥터의 스키마 감지·진화를 활성화하려면 TRUE로 지정하세요. 기본값은 FALSE입니다.

이 속성이 TRUE로 설정되면:

  • Kafka 커넥터가 만드는 새 테이블에는 테이블 매개변수 ENABLE_SCHEMA_EVOLUTION이 자동으로 TRUE로 설정됩니다.
  • 기존 테이블에는 여전히 테이블 매개변수 ENABLE_SCHEMA_EVOLUTION을 수동으로 TRUE로 설정해야 합니다.

schema.registry.url

스키마 레지스트리 서비스의 URL을 지정하세요. 기본값은 비어 있습니다.

파일 포맷에 따라 schema.registry.url은 필수이거나 선택입니다. Kafka 커넥터의 스키마 감지는 다음 시나리오 중 하나에서 지원됩니다.

  • 스키마 레지스트리는 Avro와 Protobuf에 필수입니다. 컬럼은 제공된 스키마 레지스트리에 정의된 데이터 타입으로 생성됩니다.
  • 스키마 레지스트리는 JSON에는 선택입니다. 스키마 레지스트리가 없으면 제공된 데이터를 기반으로 데이터 타입이 추론됩니다.

평소처럼 Kafka 커넥터 속성 파일에 추가 속성을 구성하세요. 자세한 내용은 Kafka 커넥터 구성을 참고하세요.

컨버터

Json, Avro, Protobuf 같은 구조적 데이터 컨버터가 지원됩니다. 다음 구조적 데이터 컨버터만 테스트했다는 점에 유의하세요.

  • io.confluent.connect.avro.AvroConverter
  • io.confluent.connect.protobuf.ProtobufConverter
  • org.apache.kafka.connect.json.JsonConverter
  • io.confluent.connect.json.JsonSchemaConverter

비구조적 데이터 컨버터는 schematization에서 지원되지 않습니다. 예:

  • org.apache.kafka.connect.converters.ByteArrayConverter
  • org.apache.kafka.connect.storage.StringConverter

Snowflake 컨버터는 Snowpipe Streaming에서 지원되지 않습니다. 일부 커스텀 데이터 컨버터는 테스트되지 않았으며 지원되지 않을 수도 있어요.

사용 참고

  • Kafka 커넥터의 스키마 감지는 스키마 레지스트리 유무와 무관하게 지원됩니다. 스키마 레지스트리를 사용하면(Avro와 Protobuf) 컬럼이 제공된 스키마 레지스트리에 정의된 데이터 타입으로 생성됩니다. 스키마 레지스트리가 없으면(JSON) 제공된 데이터를 기반으로 데이터 타입이 추론됩니다.
  • Kafka 커넥터의 스키마 진화는 다음 테이블 컬럼 수정을 지원합니다:
    • 새 컬럼 추가
    • 소스 데이터 컬럼이 없으면 NOT NULL 제약 제거
  • Kafka 커넥터가 대상 테이블을 만들면 스키마 진화가 기본적으로 활성화됩니다. 단, 기존 테이블에서 스키마 진화가 비활성화되면 Kafka 커넥터는 스키마가 일치하지 않는 행을 구성된 dead-letter queue(DLQ)로 보내려고 시도합니다.
  • JSON ARRAY는 추가 schematization이 지원되지 않습니다.
  • Snowpipe Streaming과 함께하는 Kafka 커넥터의 경우 스키마 진화는 다음 뷰와 명령에서 SchemaEvolutionRecord 출력으로 추적되지 않습니다: INFORMATION_SCHEMA COLUMNS 뷰, ACCOUNT_USAGE COLUMNS 뷰, DESCRIBE TABLE 명령, SHOW COLUMNS 명령. SchemaEvolutionRecord 출력은 항상 NULL을 보여 줍니다.

예제

다음 예제는 Snowpipe Streaming과 함께하는 Kafka 커넥터에 대해 스키마 감지·진화가 활성화되기 전과 후에 생성되는 테이블을 보여 줍니다.

-- 스키마 감지·진화가 활성화되기 전에는 다음 예제처럼 테이블이 RECORD_CONTENT와 RECORD_METADATA라는 두 VARIANT 컬럼만으로 구성됩니다.

+------+---------------------------------------------------------+---------------------------------------------------+
| Row  | RECORD_METADATA                                         | RECORD_CONTENT                                    |
|------+---------------------------------------------------------+---------------------------------------------------|
| 1    |{"CreateTime":1669074170090, "headers": {"current.iter...| "account": "ABC123", "symbol": "ZTEST", "side":...|
| 2    |{"CreateTime":1669074170400, "headers": {"current.iter...| "account": "XYZ789", "symbol": "ZABZX", "side":...|
+------+---------------------------------------------------------+---------------------------------------------------+

-- 스키마 감지·진화가 활성화된 후에는 테이블에 사용자 정의 스키마와 일치하는 컬럼이 포함됩니다. 테이블은 Kafka 커넥터가 로드하는 새 Snowpipe 스트리밍 데이터의 구조를 지원하도록 자동으로 진화할 수도 있습니다.

+------+---------------------------------------------------------+---------+--------+-------+----------+
| Row  | RECORD_METADATA                                         | ACCOUNT | SYMBOL | SIDE  | QUANTITY |
|------+---------------------------------------------------------+---------+--------+-------+----------|
| 1    |{"CreateTime":1669074170090, "headers": {"current.iter...| ABC123  | ZTEST  | BUY   | 3572     |
| 2    |{"CreateTime":1669074170400, "headers": {"current.iter...| XYZ789  | ZABZX  | SELL  | 3024     |
+------+---------------------------------------------------------+---------+--------+-------+----------+

더 알아보기 (Learn more)