Working with the 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 레코드 키 이름이 원하는 열 이름과 일치하는 경우

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

• You want a simple configuration

기본 파이프 모드를 사용하여 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 열¶

In prior versions of the connector (3.x and earlier), when the schematization feature was disabled, the connector created a destination table with two columns: RECORD_CONTENT and RECORD_METADATA. The RECORD_CONTENT column contained the entire Kafka message content in a column of type VARIANT. The RECORD_METADATA column continues to be supported but the RECORD_CONTENT column is no longer created by the connector. The same functionality can be achieved using SMT transformations (see examples later in this section). The RECORD_CONTENT key is also no longer available in PIPE transformations. For example, this PIPE definition will not work by default:

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>`_를 참조하세요.

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

The connector inspects the Snowpipe Streaming channel status before committing offsets in Kafka. If the connector detects that the rowsErrorCount property on channel has increased since the connector was started, it raises a fatal error (ERROR_5030) when errors.tolerance=none so that data issues don’t go unnoticed. To allow ingestion to continue while triaging bad rows, set errors.tolerance=all

errors.tolerance=all

외부 볼륨에 사용 권한 부여¶

Kafka 커넥터의 역할에 Apache Iceberg™ 테이블과 연결된 외부 볼륨에 대한 USAGE 권한을 부여하려면 다음 문을 실행합니다.

예를 들어, Iceberg 테이블이 kafka_external_volume 외부 볼륨을 사용하고 커넥터가 kafka_connector_role 역할을 사용하는 경우 다음 문을 실행합니다.

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kafka_external_volume TO ROLE kafka_connector_role;

수집을 위한 Apache Iceberg™ 생성¶

커넥터는 Iceberg 테이블을 자동으로 생성하지 않으며 스키마 진화를 지원하지 않습니다. 커넥터를 실행하기 전에 Iceberg 테이블을 수동으로 생성해야 합니다.

Iceberg 테이블을 생성할 때 Iceberg 데이터 타입(VARIANT 포함) 또는 :doc:`호환되는 Snowflake 타입</user-guide/tables-iceberg-data-types>`을 사용할 수 있습니다.

예를 들어, 다음 메시지를 생각해 보십시오.

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

예제 메시지에 대한 Iceberg 테이블을 생성하려면 다음 문을 사용합니다.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id number(38,0),
    name varchar,
    body_temperature number(4,2),
    approved_coffee_types array(varchar),
    animals_possessed variant,
    options object(can_walk boolean, can_talk boolean),
    date_added date
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id INT,
    name string,
    body_temperature float,
    approved_coffee_types array(string),
    animals_possessed variant,
    date_added date,
    options object(can_walk boolean, can_talk boolean),
    )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy