Snowflake High Performance connector for Kafka 작동 방식¶

항목 이름, 테이블 이름, 파이프 이름¶

구성 설정에 따라 커넥터는 대상 테이블에 대해 다른 이름을 사용합니다. 대상 테이블 이름은 항상 항목 이름에서 파생됩니다.

RECORD_METADATA 이해하기¶

커넥터는 Kafka 레코드에 대한 메타데이터가 있는 RECORD_METADATA 구조를 채웁니다. 이 메타데이터는 Snowpipe Streaming 데이터 소스를 통해 Snowflake로 전송되며, 여기서 $1:RECORD_METADATA 접근자를 사용하는 파이프 변환에 제공됩니다. RECORD_METADATA 구조는 사용자 정의 파이프 모드와 기본 파이프 모드에서 모두 사용할 수 있습니다. 해당 내용은 VARIANT 유형의 열에 저장하거나 개별 필드를 추출하여 별도의 열에 저장할 수 있습니다.

변환 및 메타데이터가 있는 예제 파이프:

CREATE PIPE ORDERS AS
COPY INTO ORDERS_TABLE
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total,
    $1:RECORD_METADATA.topic AS source_topic,
    $1:RECORD_METADATA.offset::NUMBER AS kafka_offset,
    $1:RECORD_METADATA.SnowflakeConnectorPushTime::BIGINT AS ingestion_time
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

이 예제에서:

• 파이프는 Kafka 메시지에서 특정 필드(order_id, customer_name, order_total)를 추출합니다.

• 또한 메타데이터 필드(항목, 오프셋 및 수집 타임스탬프)도 캡처합니다.

• 값은 필요에 따라 캐스팅 및/또는 변환될 수 있습니다.

메타데이터 필드를 채우는 방법¶

커넥터는 Kafka 레코드 속성 및 커넥터 구성을 기반으로 메타데이터 필드를 자동으로 채웁니다. 다음 구성 매개 변수를 사용하여 포함할 메타데이터 필드를 제어할 수 있습니다.

• ``snowflake.metadata.topic``(기본값: true) - 항목 이름 포함

• ``snowflake.metadata.offset.and.partition``(기본값: true) - 오프셋 및 파티션 포함

• ``snowflake.metadata.createtime``(기본값: true) - Kafka 레코드 타임스탬프 포함

• ``snowflake.metadata.all``(기본값: true) - 사용 가능한 모든 메타데이터 포함

``snowflake.metadata.all=true``(기본값)인 경우 모든 메타데이터 필드가 채워집니다. 개별 메타데이터 플래그를 ``false``로 설정하면 RECORD_METADATA 구조에서 해당 특정 필드를 제외합니다.

RECORD_METADATA 구조에 기본적으로 포함되는 정보는 다음과 같습니다.

RECORD_METADATA 열에 기록된 메타데이터의 양은 선택적 Kafka 구성 속성을 사용하여 구성할 수 있습니다.

필드 이름 및 값에서는 대/소문자를 구분합니다.

수집 전에 Kafka 레코드를 변환하는 방법¶

각 행이 Snowpipe Streaming으로 전달되기 전에 커넥터는 Kafka Connect 레코드 값을 ``Map<String, Object>``로 변환합니다. 해당 키는 대상 열 이름과 일치해야 합니다(또는 사용자 정의 파이프 내에서 변환될 수 있음). 커넥터가 정형화 오브젝트를 수신할 수 있도록 기본 문자열, 바이트 배열 또는 숫자는 래핑해야 합니다(예: HoistFieldSMT). 변환기는 다음 규칙을 적용합니다.

• Null 값은 삭제 표시로 처리됩니다. ``behavior.on.null.values=IGNORE``인 경우 건너뛰거나 그렇지 않은 경우 빈 JSON 오브젝트로 수집됩니다.

• 숫자 및 부울 필드는 있는 그대로 전달됩니다. 전체 자릿수가 38보다 큰 10진수 값은 Snowflake의 NUMBER 제한 내에서 유지되기 위해 문자열로 직렬화됩니다.

• byte[] 및 ByteBuffer 페이로드는 Base64로 인코딩된 문자열이므로 VARIANT 또는 VARCHAR 열에 저장합니다.

• 배열은 배열로 유지되고 중첩된 오브젝트는 중첩된 맵으로 유지됩니다. 기본 파이프를 사용하여 중첩된 데이터를 있는 그대로 랜딩하는 경우 VARIANT 열을 선언합니다.

• Snowflake 열 이름은 텍스트여야 하므로 문자열이 아닌 키가 있는 맵은 [key, value] 페어의 배열로 내보내집니다.

• 관련 메타데이터 플래그가 활성화될 때마다 레코드 헤더와 키가 ``RECORD_METADATA``에 복사됩니다.

전체 메시지 본문을 단일 열로 보존해야 하는 경우 SMTs를 사용하여 새 최상위 필드로 래핑합니다. 변환 패턴에 대해서는 :ref:`레거시 RECORD_CONTENT 열<label-kafkahp_connector_legacy_record_content_column>`을 참조하세요.

사용자 정의 파이프 모드¶

이 모드에서는 데이터 변환 및 열 매핑을 완전히 제어할 수 있습니다.

이 모드를 사용하는 경우:

• JSON 필드 이름과 다른 사용자 지정 열 이름이 필요한 경우

• 데이터 변환(형식 캐스팅, 마스킹, 필터링)을 적용해야 하는 경우

• 데이터가 열에 매핑되는 방식을 완전히 제어하려는 경우

기본 파이프 모드¶

이 모드에서 커넥터는 이름이 ``{tableName}-STREAMING``인 기본 파이프를 사용하며, kafka 레코드 필드를 이름별로 일치하는 테이블 열에 매핑합니다(대/소문자 구분 안 함).

이 모드를 사용하는 경우:

• kafka 레코드 키 이름이 원하는 열 이름과 일치하는 경우

• 사용자 지정 데이터 변환이 필요하지 않은 경우

• 간소화된 구성을 원하는 경우

기본 파이프 모드를 사용하여 kafka 레코드 키를 테이블 열에 매핑하기

기본 파이프 모드를 사용할 때 커넥터는 이름이 ``{tableName}-STREAMING``인 기본 파이프를 사용하며, 대/소문자를 구분하지 않는 일치를 통해 콘텐츠의 첫 번째 수준 키를 테이블 열에 직접 매핑합니다.

기본 파이프 모드 사용하기 - 예제¶

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  "@&$#* includes special characters": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  record_metadata VARIANT,
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  "has cat" BOOLEAN,
  "!@&$#* includes special characters" BOOLEAN,
  skills VARIANT,
  family VARIANT
);

사용자 정의 파이프 모드 사용하기 - 예제¶

이 예제에서는 사용자 지정 데이터 변환으로 사용자 정의 파이프를 구성하고 사용하는 방법을 보여줍니다.

CREATE TABLE ORDERS (
  order_id NUMBER,
  customer_name VARCHAR,
  order_total NUMBER,
  order_date TIMESTAMP_NTZ,
  source_topic VARCHAR
);

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total::NUMBER,
    $1:order_date::TIMESTAMP_NTZ,
    $1:RECORD_METADATA.topic
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

snowflake.topic2table.map=kafka-orders-topic:ORDERS

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  " @&$#* has Łułósżź": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  has_cat BOOLEAN,
  weird_field_name BOOLEAN,
  skills VARIANT,
  family VARIANT
);

CREATE PIPE PERSON_DATA AS
COPY INTO PERSON_DATA
FROM (
  SELECT
    $1:city,
    $1:age,
    $1:married,
    $1['has cat'] AS has_cat,
    $1[' @&$#* has Łułósżź'] AS weird_field_name,
    $1:skills,
    $1:family
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

명시적 항목-테이블 매핑¶

snowflake.topic2table.map 매개 변수를 구성할 때 커넥터는 명시적 매핑 모드에서 작동합니다. 이 모드에서는 다음을 수행할 수 있습니다.

• 여러 Kafka 항목을 단일 Snowflake 테이블에 매핑

• 항목 이름과 다른 사용자 지정 테이블 이름 사용

• 여러 항목과 일치하도록 정규식 패턴 적용

구성 형식:

snowflake.topic2table.map 매개 변수는 쉼표로 구분된 항목-테이블 매핑 목록을 다음 형식으로 허용합니다.

topic1:table1,topic2:table2,topic3:table3

구성 예제:

직접 항목 매핑

snowflake.topic2table.map=orders:ORDER_TABLE,customers:CUSTOMER_TABLE

정규식 패턴 일치

snowflake.topic2table.map=.*_cat:CAT_TABLE,.*_dog:DOG_TABLE

이 구성은 _cat``(예: ``orange_cat, calico_cat)으로 끝나는 모든 항목을 CAT_TABLE 테이블에 매핑하고 _dog``로 끝나는 모든 항목을 ``DOG_TABLE 테이블에 매핑합니다.

하나의 테이블에 여러 항목

snowflake.topic2table.map=topic1:shared_table,topic2:shared_table,topic3:other_table

이 구성은 topic1 및 ``topic2``를 모두 ``shared_table``에 매핑하고 ``topic3``은 ``other_table``에 매핑합니다.

레거시 RECORD_CONTENT 열¶

이전 버전의 커넥터에서는 스키마화 기능이 비활성화되었을 때 커넥터가 RECORD_CONTENT 및 RECORD_METADATA의 두 열이 있는 대상 테이블을 생성했습니다. RECORD_CONTENT 열에는 유형 VARIANT의 열에 전체 Kafka 메시지 내용이 포함되어 있습니다. RECORD_METADATA 열은 계속 지원되지만 RECORD_CONTENT 열은 커넥터에서 더 이상 생성되지 않습니다. SMT 변환을 사용하여 동일한 기능을 구현할 수 있습니다(이 섹션의 뒷부분에 있는 예제 참조). 또한 RECORD_CONTENT 키는 PIPE 변환에서 더 이상 사용할 수 없습니다. 예를 들어, 이 PIPE 정의는 기본적으로 작동하지 않습니다.

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:RECORD_CONTENT
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

단일 열에 전체 Kafka 메시지 내용을 저장해야 하거나 PIPE 변환의 전체 Kafka 메시지 내용을 처리해야 하는 경우, 전체 Kafka 메시지 내용을 원하는 사용자 지정 필드로 래핑하는 다음 SMT 변환을 사용할 수 있습니다.

transforms=wrapKafkaMessageContent
transforms.wrapKafkaMessageContent.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.wrapKafkaMessageContent.field=your_top_level_field_name

이 변환은 전체 Kafka 메시지 내용을 your_top_level_field_name``이라는 사용자 지정 필드로 래핑합니다. 그런 다음 PIPE 변환의 ``$1:your_top_level_field_name 접근자를 사용하여 전체 Kafka 메시지 내용에 액세스할 수 있습니다.

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:your_top_level_field_name
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

또는 기본 파이프를 사용하여 전체 메타데이터와 콘텐츠를 모두 단일 테이블에 저장하려면 사용자 지정 파이프를 생성하지 마세요. 대신, RECORD_CONTENT 및 ``your_top_level_field_name``이라는 두 열이 있는 테이블만 생성합니다.

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

HoistField$Value 변환에 대해 자세히 알아보려면 `Kafka 설명서<https://kafka.apache.org/39/documentation.html#connect_transforms>`_를 참조하세요.

스트리밍 채널 오류 및 데드 레터 큐 처리하기¶

4.0.0-rc4 버전에서 커넥터는 오프셋을 커밋하기 전에 Snowpipe Streaming 채널 상태를 검사합니다. Snowflake가 거부된 행을 보고하는 경우(rowsErrorCount > 0), the connector now raises a fatal error (ERROR_5030) when errors.tolerance=none so that data issues cannot go unnoticed. To allow ingestion to continue while you triage bad rows, set errors.tolerance=all

errors.tolerance=all

커넥터의 내결함성 제한 사항¶

Kafka 항목은 저장 공간 또는 보존 시간에 대한 제한으로 구성될 수 있습니다.

• 시스템이 보존 시간 이상 동안 오프라인 상태로 유지되면 만료된 레코드는 로드되지 않습니다. 유사하게 Kafka의 저장소 공간 제한이 초과되면 일부 메시지가 전달되지 않습니다.

• Kafka 항목의 메시지가 삭제 또는 업데이트되는 경우 해당 변경 사항이 Snowflake 테이블에 반영되지 않을 수 있습니다.