Snowflake High Performance connector for Kafka: Install and configure¶

Installing the connector for Confluent¶

오픈 소스 Apache Kafka용 커넥터 설치하기¶

이 섹션에서는 오픈 소스 Apache Kafka용 Kafka 커넥터를 설치 및 구성하기 위한 지침을 제공합니다.

분산형 모드¶

Kafka 구성 파일(예: <path>/<config_file>.json)을 생성합니다. 모든 커넥터 구성 정보로 파일을 채웁니다. 파일은 JSON 형식이어야 합니다.

Sample configuration file

{
  "name":"XYZCompanySensorData",
  "config":{
      "connector.class": "com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector",
      "tasks.max": "1",
      "snowflake.topic2table.map": "topic1:table_1,topic2:table_2",
      "snowflake.url.name": "myorganization-myaccount.snowflakecomputing.com:443",
      "snowflake.warehouse.name": "WH",
      "snowflake.private.key": "-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n",
      "snowflake.schema.name": "MY_SCHEMA",
      "snowflake.database.name": "MY_DATABASE",
      "snowflake.role.name": "MY_ROLE",
      "snowflake.user.name": "MY_USER",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "errors.log.enable": "true",
      "topics": "topic1,topic2",
      "value.converter.schemas.enable": "false",
      "errors.tolerance": "none"
      }
}

독립 실행형 모드¶

구성 파일(예 <kafka_dir>/config/SF_connect.properties)을 생성합니다. 모든 커넥터 구성 정보로 파일을 채웁니다.

Sample configuration file

connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=1
snowflake.topic2table.map=topic1:table_1,topic2:table_2
snowflake.url.name=myorganization-myaccount.snowflakecomputing.com:443
snowflake.warehouse.name=WH
snowflake.private.key=-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n
snowflake.schema.name=MY_SCHEMA
snowflake.database.name=MY_DATABASE
snowflake.role.name=MY_ROLE
snowflake.user.name=MY_USER
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
errors.log.enable=true
topics=topic1,topic2
name=XYZCompanySensorData
value.converter.schemas.enable=false
errors.tolerance=none

필수 속성¶

name

애플리케이션 이름입니다. 이 이름은 고객이 사용하는 모든 Kafka 커넥터에서 고유해야 합니다. 이 이름은 따옴표로 묶이지 않은 유효한 Snowflake 식별자여야 합니다. 유효한 식별자에 대한 내용은 식별자 요구 사항 을 참조하십시오.

connector.class

com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector

topics

항목의 쉼표로 구분된 목록입니다. 기본적으로 Snowflake는 테이블 이름과 동일한 항목 이름을 가정합니다. 테이블 이름이 항목 이름과 동일하지 않은 경우 선택적 topic2table.map 매개 변수(아래)를 사용하여 항목 이름에서 테이블 이름으로 매핑을 지정합니다. 이 테이블 이름은 따옴표로 묶이지 않은 유효한 Snowflake 식별자여야 합니다. 유효한 테이블 이름에 대한 내용은 식별자 요구 사항 을 참조하십시오.

참고

둘 다가 아닌 topics 또는 topics.regex 중 1개 가 필수입니다.

topics.regex

Snowflake 테이블에 로드할 메시지가 포함된 항목을 지정하는 정규식(“regex”)입니다. 커넥터는 정규식과 일치하는 모든 항목 이름에서 데이터를 로드합니다. 정규식은 반드시 Java 정규식의 규칙을 따라야 합니다(즉, java.util.regex.Pattern과 호환되어야 함). 구성 파일에는 topics 또는 topics.regex 중 1개 가 포함되어야 하며 둘 다 포함되지 않아야 합니다.

snowflake.url.name

Snowflake 계정에 액세스하기 위한 URL입니다. 이 URL에는 반드시 계정 식별자 가 포함되어야 합니다. 프로토콜(https://) 및 포트 번호는 선택 사항입니다.

snowflake.user.name

Snowflake 계정의 사용자 로그인 이름입니다.

snowflake.role.name

커넥터가 테이블에 데이터를 삽입하는 데 사용할 역할의 이름입니다.

snowflake.private.key

사용자 인증에서 사용되는 개인 키입니다. 키만 포함되어야 하며 헤더 또는 푸터는 포함되지 않아야 합니다. 키가 여러 라인으로 분할되는 경우에는 줄 바꿈을 제거합니다. 암호화되지 않은 키를 제공하거나 암호화된 키를 제공하고 snowflake.private.key.passphrase 매개 변수를 제공하여 Snowflake가 키의 암호를 해독할 수 있도록 할 수 있습니다. :emph:` 매개 변수 값이 암호화된 경우에만``snowflake.private.key` 이 매개 변수를 사용하십시오. :doc:`/user-guide/key-pair-auth`의 지침에 따라 암호화된 개인 키의 암호를 해독합니다.

참고

label-optional_properties`의 :samp:`snowflake.private.key.passphrase 섹션을 참조하세요.

snowflake.database.name

행을 삽입할 테이블이 포함된 데이터베이스의 이름입니다.

snowflake.schema.name

행을 삽입할 테이블이 포함된 스키마의 이름입니다.

header.converter

레코드가 Avro로 형식이 지정되고 헤더가 포함된 경우에만 필요합니다. 값은 "org.apache.kafka.connect.storage.StringConverter" 입니다.

key.converter

Kafka 레코드의 키 변환기(예: "org.apache.kafka.connect.storage.StringConverter")입니다. 이 속성은 Kafka 커넥터에서 사용되지 않지만 Kafka Connect 플랫폼에서는 필요합니다.

현재 제한 사항은 Kafka 커넥터 제한 사항 을 참조하십시오.

value.converter

커넥터는 표준 Kafka 커뮤니티 변환기를 지원합니다. 데이터 형식에 따라 적절한 변환기를 선택합니다.

• JSON레코드의 경우: "org.apache.kafka.connect.json.JsonConverter"

• Schema Registry가 있는 Avro 레코드의 경우: "io.confluent.connect.avro.AvroConverter"

현재 제한 사항은 Kafka 커넥터 제한 사항 을 참조하십시오.

선택적 속성¶

snowflake.private.key.passphrase

이 매개 변수의 값이 비어 있지 않은 경우 커넥터는 이 구문을 사용하여 개인 키의 암호를 해독하려고 시도합니다.

tasks.max

작업 수이며, 일반적으로 Kafka Connect 클러스터의 작업자 노드에 있는 CPU 코어 수와 동일합니다. 최상의 성능을 얻으려면 작업 수를 Kafka 파티션의 총 수와 동일하게 설정하되 CPU 코어 수를 초과하지 않도록 설정하는 것이 좋습니다. 작업 수가 많으면 메모리 사용량이 증가하고 재밸런싱이 자주 발생할 수 있습니다.

snowflake.topic2table.map

이 선택적 매개 변수를 사용하면 사용자가 테이블에 매핑할 항목을 지정할 수 있습니다. 각 항목과 테이블 이름은 콜론으로 구분해야 합니다(아래 예 참조). 이 테이블 이름은 따옴표로 묶이지 않은 유효한 Snowflake 식별자여야 합니다. 유효한 테이블 이름에 대한 내용은 식별자 요구 사항 을 참조하십시오. 항목 구성에서는 topics.regex 를 사용할 때와 마찬가지로 정규식을 사용하여 항목을 정의할 수 있습니다. 정규 식은 모호할 수 없습니다. 일치하는 항목은 단 하나의 대상 테이블과만 일치해야 합니다.

예:

topics="topic1,topic2,topic5,topic6"
snowflake.topic2table.map="topic1:low_range,topic2:low_range,topic5:high_range,topic6:high_range"

Copy

다음과 같이 작성할 수 있습니다:

topics.regex="topic[0-9]"
snowflake.topic2table.map="topic[0-4]:low_range,topic[5-9]:high_range"

Copy

value.converter.schema.registry.url

형식이 Avro이고 Schema Registry Service를 사용하는 경우 이 속성은 Schema Registry Service의 URL이어야 합니다. 그렇지 않으면 이 필드를 비워둬야 합니다.

value.converter.break.on.schema.registry.error

스키마 레지스트리 서비스에서 Avro 데이터를 로드하는 경우 이 속성은 스키마 ID를 가져오는 동안 오류가 발생하면 Kafka 커넥터가 레코드 사용을 중지해야 하는지 여부를 결정합니다. 기본값은 false 입니다. 이 동작을 활성화하려면 값을 true 로 설정합니다.

jvm.proxy.host

Snowflake Kafka Connector가 프록시 서버를 통해 Snowflake에 액세스할 수 있도록 하려면 이 매개 변수를 설정하여 해당 프록시 서버의 호스트를 지정합니다.

jvm.proxy.port

Snowflake Kafka Connector가 프록시 서버를 통해 Snowflake에 액세스할 수 있도록 하려면 이 매개 변수를 설정하여 해당 프록시 서버의 포트를 지정합니다.

snowflake.streaming.max.client.lag

Specifies how often the connector flushes the data to Snowflake, in seconds.

값:

• 최소: 1 초

• 최대: 600 초

기본값:

:code:`1`초

jvm.proxy.username

프록시 서버로 인증하는 사용자 이름입니다.

jvm.proxy.password

프록시 서버로 인증하는 사용자 이름의 비밀번호입니다.

snowflake.jdbc.map

예: "snowflake.jdbc.map": "networkTimeout:20,tracing:WARNING"

추가 JDBC 속성(JDBC 드라이버 연결 매개 변수 참조 참조)은 유효성이 검사되지 않습니다. 이러한 추가 속성은 유효성이 검사되지 않으며 필수 속성(예: jvm.proxy.xxx, snowflake.user.name, snowflake.private.key, snowflake.schema.name) 대신 재정의하거나 사용해서는 안 됩니다.

다음 조합 중 하나를 지정합니다.

• JDBC_TRACE 환경 변수와 함께 tracing 속성

• snowflake.database.name 와 함께 database 속성

모호한 동작이 발생하며 동작은 JDBC 드라이버에 의해 결정됩니다.

value.converter.basic.auth.credentials.source

Avro 데이터 타입을 사용 중이고 Kafka 스키마 레지스트리에 대한 보안 액세스가 필요한 경우 이 매개 변수를 “USER_INFO” 문자열로 설정하고 아래 설명된 value.converter.basic.auth.user.info 매개 변수를 설정합니다. 그렇지 않으면, 이 매개 변수를 생략합니다.

value.converter.basic.auth.user.info

Avro 데이터 타입을 사용 중이고 Kafka 스키마 레지스트리에 대한 보안 액세스가 필요한 경우 위의 설명과 같이 이 매개 변수를 “<사용자_ID>:<비밀번호>” 문자열로 설정하고 value.converter.basic.auth.credentials.source 매개 변수를 설정합니다. 그렇지 않으면, 이 매개 변수를 생략합니다.

snowflake.metadata.createtime

값이 FALSE로 설정되면 RECORD_METADATA 열의 메타데이터에서 CreateTime 속성 값이 생략됩니다. 기본값은 TRUE입니다.

snowflake.metadata.topic

값이 FALSE로 설정되면 RECORD_METADATA 열의 메타데이터에서 topic 속성 값이 생략됩니다. 기본값은 TRUE입니다.

snowflake.metadata.offset.and.partition

값이 FALSE로 설정되면 RECORD_METADATA 열의 메타데이터에서 Offset 및 Partition 속성 값이 생략됩니다. 기본값은 TRUE입니다.

snowflake.metadata.all

값이 FALSE로 설정되면 RECORD_METADATA 열의 메타데이터가 완전히 비어 있습니다. 기본값은 TRUE입니다.

transforms

Kafka 커넥터가 발견한 삭제 표시 레코드를 건너뛰고 대상 테이블에 로드하지 않도록 지정합니다. 삭제 표시 레코드는 전체 값 필드가 null인 레코드로 정의됩니다.

속성 값을 "tombstoneHandlerExample" 로 설정합니다.

참고

Kafka 커뮤니티 변환기(즉, value.converter 속성 값)에만 이 속성(예: org.apache.kafka.connect.json.JsonConverter 또는 org.apache.kafka.connect.json.AvroConverter)을 사용합니다. Snowflake 변환기로 삭제 표시 레코드 처리를 관리하려면 behavior.on.null.values 속성을 대신 사용합니다.

transforms.tombstoneHandlerExample.type

transforms 속성을 설정할 때 필수입니다.

속성 값을 "io.confluent.connect.transforms.TombstoneHandler" 로 설정합니다.

behavior.on.null.values

Kafka 커넥터가 삭제 표시 레코드를 처리하는 방법을 지정합니다. 삭제 표시 레코드는 전체 값 필드가 null인 레코드로 정의됩니다. Snowpipe 의 경우 이 속성은 Kafka 커넥터 버전 1.5.5 이상에서 지원됩니다. Snowpipe Streaming 의 경우 이 속성은 Kafka 커넥터 버전 2.1.0 이상에서 지원됩니다.

이 속성에서 지원되는 값은 다음과 같습니다.

DEFAULT

Kafka 커넥터에서 삭제 표시 레코드가 발생하면 내용 열에 빈 JSON 문자열이 삽입됩니다.

IGNORE

Kafka 커넥터가 삭제 표시 레코드를 건너뛰고 해당 레코드에 대한 행을 삽입하지 않습니다.

기본값은 DEFAULT 입니다.

참고

삭제 표시 레코드 수집은 수집 방법에 따라 다릅니다.

• Snowpipe의 경우 Kafka 커넥터는 Snowflake 변환기만 사용합니다. Kafka 커뮤니티 변환기를 사용하여 삭제 표시 레코드 처리를 관리하려면 transform 및 transforms.tombstoneHandlerExample.type 속성을 대신 사용해야 합니다.

• Snowpipe Streaming의 경우 Kafka 커넥터는 커뮤니티 변환기만 사용합니다.

Kafka 브로커에 전송된 레코드는 Kafka 커넥터에 의해 삭제되어 오프셋이 누락되므로 NULL이면 안 됩니다. 오프셋이 누락되면 특정 사용 사례에서 Kafka 커넥터가 손상됩니다. NULL 레코드 대신 삭제 표식 레코드를 사용하는 것이 좋습니다.

키 페어 인증 및 키 순환 사용하기¶

Kafka 커넥터는 사용자 이름 및 비밀번호 인증 대신 키 페어 인증을 사용합니다. 이 인증 메서드에는 2048비트(최소) RSA 키 페어가 필요합니다. OpenSSL을 사용하여 공개-개인 키 페어를 생성합니다. 공개 키는 구성 파일에 정의된 Snowflake 사용자에게 할당됩니다.

이 페이지의 키 페어 인증 작업과 키 페어 순환 작업을 완료한 후, 이 항목의 뒷부분에 있는 :ref:`label-kafkahp_externalize_secrets`에 대한 권장 사항을 평가합니다.

공개/개인 키 페어를 구성하려면:

1. 터미널 창의 명령줄에서 개인 키를 생성합니다.

   개인 키의 암호화된 버전 또는 암호화되지 않은 버전을 생성할 수 있습니다.

   참고

   Kafka 커넥터는 연방 정보 처리 표준(140-2)(즉, FIPS 140-2) 요구 사항을 충족하는 것으로 검증된 암호화 알고리즘을 지원합니다. 자세한 내용은 FIPS 140-2 를 참조하십시오.

   암호화되지 않은 버전을 생성하려면 다음 명령을 사용합니다.

   $ openssl genrsa -out rsa_key.pem 2048
   

   Copy

   암호화된 버전을 생성하려면 다음 명령을 사용합니다.

   $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 <algorithm> -inform PEM -out rsa_key.p8
   

   Copy

   여기서 <알고리즘> 은 FIPS 140-2 호환 암호화 알고리즘입니다.

   예를 들어, AES 256을 암호화 알고리즘으로 지정하려면:

   $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 aes256 -inform PEM -out rsa_key.p8
   

   Copy

   암호화된 버전의 개인 키를 생성하는 경우에는 암호 구문을 기록해 두십시오. 나중에 Kafka 구성 파일의 snowflake.private.key.passphrase 속성에 암호 구문을 지정합니다.

   샘플 PEM 개인 키

   -----BEGIN ENCRYPTED PRIVATE KEY-----
   MIIE6TAbBgkqhkiG9w0BBQMwDgQILYPyCppzOwECAggABIIEyLiGSpeeGSe3xHP1
   wHLjfCYycUPennlX2bd8yX8xOxGSGfvB+99+PmSlex0FmY9ov1J8H1H9Y3lMWXbL
   ...
   -----END ENCRYPTED PRIVATE KEY-----
   

   Copy

2. 명령줄에서 개인 키를 참조하여 공개 키를 생성합니다.

   개인 키가 암호화되고 이름이 rsa_key.p8 인 파일에 포함된 것으로 가정하고 다음 명령을 사용합니다.

   $ openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
   

   Copy

   샘플 PEM 공개 키

   -----BEGIN PUBLIC KEY-----
   MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy+Fw2qv4Roud3l6tjPH4
   zxybHjmZ5rhtCz9jppCV8UTWvEXxa88IGRIHbJ/PwKW/mR8LXdfI7l/9vCMXX4mk
   ...
   -----END PUBLIC KEY-----
   

   Copy

3. 공개 및 개인 키 파일을 로컬 디렉터리로 복사하여 저장합니다. 파일 경로를 기록해 둡니다. 개인 키는 PKCS#8(공개 키 암호화 표준) 형식을 사용하여 저장되며 이전 단계에서 지정한 암호 구문을 사용하여 암호화되지만, 파일은 운영 체제에서 제공하는 파일 메커니즘을 사용하여 무단 액세스로부터 보호되어야 합니다. 파일을 사용하지 않을 때 파일을 보호하는 것은 사용자의 책임입니다.

4. Snowflake에 로그인합니다. ALTER USER 를 사용하여 Snowflake 사용자에게 공개 키를 할당합니다.

   For example:

   ALTER USER jsmith SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';
   

   Copy

   참고

   • 보안 관리자(즉, SECURITYADMIN 역할의 사용자) 이상만 사용자를 변경할 수 있습니다.

   • SQL 문에서 공개 키 헤더와 푸터를 제외합니다.

   DESCRIBE USER 를 사용하여 사용자의 공개 키 지문을 확인합니다.

   DESC USER jsmith;
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   | property                      | value                                               | default | description                                                                   |
   |-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------|
   | NAME                          | JSMITH                                              | null    | Name                                                                          |
   ...
   ...
   | RSA_PUBLIC_KEY_FP             | SHA256:nvnONUsfiuycCLMXIEWG4eTp4FjhVUZQUQbNpbSHXiA= | null    | Fingerprint of user's RSA public key.                                         |
   | RSA_PUBLIC_KEY_2_FP           | null                                                | null    | Fingerprint of user's second RSA public key.                                  |
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   

   Copy

   참고

   RSA_PUBLIC_KEY_2_FP 속성에 대한 설명은 키-페어 순환 구성하기 에서 제공됩니다.

5. 전체 개인 키를 복사하여 구성 파일의 snowflake.private.key 필드에 붙여넣습니다. 파일을 저장합니다.

분산형 모드¶

터미널 창에서 다음 명령을 실행합니다.

curl -X POST -H "Content-Type: application/json" --data @<path>/<config_file>.json http://localhost:8083/connectors

독립 실행형 모드¶

터미널 창에서 다음 명령을 실행합니다.

<kafka_dir>/bin/connect-standalone.sh <kafka_dir>/<path>/connect-standalone.properties <kafka_dir>/config/SF_connect.properties