Snowflake High Performance connector for Kafka: Install and configure¶

Installing the connector for Confluent¶

Instalação do conector para o Apache Kafka de código aberto¶

Esta seção fornece instruções para instalar e configurar o conector Kafka para o Apache Kafka de código aberto.

Modo distribuído¶

Crie o arquivo de configuração do Kafka, por exemplo, <path>/<config_file>.json. Preencha o arquivo com todas as informações de configuração do conector. O arquivo deve estar no formato 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"
      }
}

Modo autônomo¶

Crie um arquivo de configuração, por exemplo, <kafka_dir>/config/SF_connect.properties. Preencha o arquivo com todas as informações de configuração do conector.

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

Propriedades obrigatórias¶

name

Nome do aplicativo. Isto deve ser único em todos os conectores Kafka utilizados pelo cliente. Este nome deve ser um identificador válido do Snowflake, sem aspas. Para obter mais informações sobre identificadores válidos, consulte Requisitos para identificadores.

connector.class

com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector

topics

Lista de tópicos separados por vírgula. Por padrão, o Snowflake considera que o nome da tabela é o mesmo que o nome do tópico. Se o nome da tabela não for o mesmo que o nome do tópico, então use o parâmetro opcional topic2table.map (abaixo) para especificar o mapeamento do nome do tópico para o nome da tabela. Este nome de tabela deve ser um identificador válido do Snowflake, sem aspas. Para obter mais informações sobre nomes válidos de tabelas, consulte Requisitos para identificadores.

Nota

Ou topics ou topics.regex é necessário; não ambos.

topics.regex

Esta é uma expressão regular (“regex”) que especifica os tópicos que contêm as mensagens a serem carregadas nas tabelas do Snowflake. O conector carrega dados de qualquer nome de tópico que coincida com a expressão regex. A expressão regex deve seguir as regras para expressões regulares Java (isto é, ser compatível com java.util.regex.Pattern). O arquivo de configuração deve conter ou topics ou topics.regex, não ambos.

snowflake.url.name

O URL para acessar sua conta Snowflake. Este URL deve incluir seu identificador da conta. Observe que o protocolo (https://) e o número da porta são opcionais.

snowflake.user.name

Nome de login do usuário para a conta Snowflake.

snowflake.role.name

O nome da função que o conector usará para inserir dados na tabela.

snowflake.private.key

A chave privada para autenticar o usuário. Inclua apenas a chave, não o cabeçalho ou rodapé. Se a chave estiver dividida em várias linhas, remova as quebras de linha. Você pode fornecer uma chave não criptografada, ou pode fornecer uma chave criptografada e fornecer o parâmetro snowflake.private.key.passphrase para permitir que o Snowflake descriptografe a chave. Use este parâmetro se e somente se o valor do parâmetro snowflake.private.key estiver criptografado. Ele descriptografa chaves privadas que foram criptografadas de acordo com as instruções em Autenticação de pares de chaves e rotação de pares de chaves.

Nota

Consulte também snowflake.private.key.passphrase em Propriedades opcionais.

snowflake.database.name

O nome do banco de dados que contém a tabela onde inserir as linhas.

snowflake.schema.name

O nome do esquema que contém a tabela onde inserir as linhas.

header.converter

Obrigatório somente se os registros forem formatados em Avro e incluírem um cabeçalho. O valor é "org.apache.kafka.connect.storage.StringConverter".

key.converter

Este é o conversor de chave do registro Kafka (por exemplo, "org.apache.kafka.connect.storage.StringConverter"). Não é usado pelo conector Kafka, mas é exigido pela plataforma Kafka Connect.

Consulte Limitações do conector Kafka para as limitações atuais.

value.converter

O conector é compatível com os conversores padrão da comunidade Kafka. Escolha o conversor apropriado com base em seu formato de dados:

• Para registros JSON: "org.apache.kafka.connect.json.JsonConverter"

• Para registros Avro com Schema Registry: "io.confluent.connect.avro.AvroConverter"

Consulte Limitações do conector Kafka para as limitações atuais.

Propriedades opcionais¶

snowflake.private.key.passphrase

Se o valor desse parâmetro não estiver vazio, o conector usará essa frase para tentar descriptografar a chave privada.

tasks.max

Número de tarefas, geralmente o mesmo que o número de núcleos de CPU dos nós de trabalhadores no cluster Kafka Connect. Para obter o melhor desempenho, a Snowflake recomenda definir o número de tarefas igual ao número total de partições Kafka, mas não excedendo o número de núcleos da CPU. Um alto número de tarefas pode resultar em maior consumo de memória e redistribuições frequentes.

snowflake.topic2table.map

Este parâmetro opcional permite que um usuário especifique quais tópicos devem ser mapeados para quais tabelas. Cada tópico e seu nome de tabela devem ser separados por dois pontos (ver exemplo abaixo). Este nome de tabela deve ser um identificador válido do Snowflake, sem aspas. Para obter mais informações sobre nomes válidos de tabelas, consulte Requisitos para identificadores. A configuração do tópico permite o uso de expressões regulares para definir tópicos, assim como o uso de topics.regex. As expressões regulares não podem ser ambíguas — qualquer tópico correspondente deve corresponder apenas a uma única tabela de destino.

Exemplo:

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

Copy

poderia ser escrito como:

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

Copy

value.converter.schema.registry.url

Se o formato for Avro e você estiver usando um Schema Registry Service, deve ser o URL do Schema Registry Service. Caso contrário, este campo deve estar vazio.

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

Se carregar dados Avro do Schema Registry Service, esta propriedade determina se o conector Kafka deve parar de consumir registros se encontrar um erro ao buscar a identificação do esquema. O valor padrão é false. Defina o valor como true para permitir este comportamento.

jvm.proxy.host

Para permitir que o conector do Kafka para Snowflake possa acessar o Snowflake através de um servidor proxy, defina este parâmetro para especificar o host desse servidor proxy.

jvm.proxy.port

Para permitir que o conector do Kafka para Snowflake possa acessar o Snowflake através de um servidor proxy, defina este parâmetro para especificar a porta desse servidor proxy.

snowflake.streaming.max.client.lag

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

Valores:

• Mínimo: 1 segundo

• Máximo: 600 segundos

Padrão:

1 segundo

jvm.proxy.username

Nome de usuário que se autentica com o servidor proxy.

jvm.proxy.password

Senha para o nome de usuário que se autentica com o servidor proxy.

snowflake.jdbc.map

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

Propriedades JDBC adicionais (consulte Referência dos parâmetros de conexão do driver JDBC) não são validadas. Essas propriedades adicionais não são validadas e não devem substituir nem ser usadas em vez de propriedades necessárias, como: jvm.proxy.xxx, snowflake.user.name, snowflake.private.key, snowflake.schema.name etc.

Especificar qualquer uma das seguintes combinações:

• Propriedade tracing junto com variável JDBC_TRACE env

• Propriedade database junto com snowflake.database.name

Resultará em um comportamento ambíguo, e o comportamento será determinado pelo Driver JDBC.

value.converter.basic.auth.credentials.source

Se você estiver usando o formato de dados Avro e precisar de acesso seguro ao registro do esquema Kafka, defina este parâmetro com a cadeia de caracteres “USER_INFO“, e defina o parâmetro value.converter.basic.auth.user.info descrito abaixo. Caso contrário, omita este parâmetro.

value.converter.basic.auth.user.info

Se você estiver usando o formato de dados Avro e precisar de acesso seguro ao registro do esquema Kafka, defina este parâmetro com a cadeia de caracteres “<ID_usuário>:<senha>” e defina o parâmetro value.converter.basic.auth.credentials.source descrito acima. Caso contrário, omita este parâmetro.

snowflake.metadata.createtime

Se o valor for definido como FALSE, o valor da propriedade CreateTime será omitido dos metadados na coluna RECORD_METADATA. O valor padrão é TRUE.

snowflake.metadata.topic

Se o valor for definido como FALSE, o valor da propriedade topic será omitido dos metadados na coluna RECORD_METADATA. O valor padrão é TRUE.

snowflake.metadata.offset.and.partition

Se o valor for definido como FALSE, os valores de propriedade Offset e Partition são omitidos dos metadados na coluna RECORD_METADATA. O valor padrão é TRUE.

snowflake.metadata.all

Se o valor for definido como FALSE, os metadados na coluna RECORD_METADATA estarão completamente vazios. O valor padrão é TRUE.

transforms

Especifique para ignorar os registros de marca de exclusão encontrados pelo conector Kafka e não carregá-los na tabela de destino. Um registro de marca de exclusão é definido como um registro onde todo o campo de valor é nulo.

Ajuste o valor da propriedade para "tombstoneHandlerExample".

Nota

Use esta propriedade somente com os conversores da comunidade Kafka (ou seja, valor da propriedade value.converter) (por exemplo, org.apache.kafka.connect.json.JsonConverter ou org.apache.kafka.connect.json.AvroConverter). Para gerenciar o manuseio de registros de marcas de exclusão com os conversores do Snowflake, use a propriedade behavior.on.null.values em seu lugar.

transforms.tombstoneHandlerExample.type

Necessário ao definir a propriedade transforms.

Defina o valor da propriedade como "io.confluent.connect.transforms.TombstoneHandler"

behavior.on.null.values

Especifique como o conector Kafka deve lidar com os registros de marca de exclusão. Um registro de marca de exclusão é definido como um registro onde todo o campo de valor é nulo. Para Snowpipe, esta propriedade é compatível com a versão 1.5.5 do conector Kafka e posterior. Para Snowpipe Streaming, essa propriedade é compatível com o conector Kafka versão 2.1.0 e posterior.

Esta propriedade suporta os seguintes valores:

DEFAULT

Quando o conector Kafka encontra um registro de marca de exclusão, insere uma cadeia de caracteres JSON vazia na coluna de conteúdo.

IGNORE

O conector Kafka ignora registros de marca de exclusão e não insere linhas para estes registros.

O valor padrão é DEFAULT.

Nota

A ingestão de registros Tombstone varia de acordo com os métodos de ingestão:

• Para Snowpipe, o conector Kafka usa apenas conversores Snowflake. Para gerenciar o manuseio de registros de marcas de exclusão com os conversores da comunidade Kafka, use as propriedades transform e transforms.tombstoneHandlerExample.type.

• Para Snowpipe Streaming, o conector Kafka usa apenas conversores de comunidade.

Os registros enviados aos brokers do Kafka não devem ser NULL porque esses registros serão eliminados pelo conector Kafka, resultando em offsets ausentes. Os offsets ausentes destruirão o conector Kafka em casos de uso específicos. É recomendável usar registros de marca de exclusão em vez de registros NULL.

Uso da autenticação de par de chaves e rotação de chaves¶

O conector Kafka depende da autenticação de par de chaves, e não da autenticação de nome de usuário e senha. Este método de autenticação requer um par de chaves RSA de 2048 bits (mínimo). Gere o par de chaves pública-privada usando OpenSSL. A chave pública é atribuída ao usuário do Snowflake definido no arquivo de configuração.

Depois de concluir as tarefas de autenticação do par de chaves nesta página e as tarefas para rotação do par de chaves, avalie a recomendação em Externalização de segredos, mais adiante neste tópico.

Para configurar o par de chaves pública/privada:

1. A partir da linha de comando em uma janela terminal, gere uma chave privada.

   Você pode gerar tanto uma versão criptografada quanto uma versão não criptografada da chave privada.

   Nota

   O conector Kafka suporta algoritmos de criptografia que são validados para atender aos requisitos do Federal Information Processing Standard (140-2) (ou seja, FIPS 140-2). Para obter mais informações, consulte FIPS 140-2.

   Para gerar uma versão não criptografada, use o seguinte comando:

   $ openssl genrsa -out rsa_key.pem 2048
   

   Copy

   Para gerar uma versão criptografada, use o seguinte comando:

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

   Copy

   Onde o <algoritmo> é um algoritmo de criptografia compatível com FIPS 140-2.

   Por exemplo, para especificar AES 256 como o algoritmo de criptografia:

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

   Copy

   Se você gerar uma versão criptografada da chave privada, registre a senha. Mais tarde, você especificará a senha na propriedade snowflake.private.key.passphrase no arquivo de configuração do Kafka.

   Exemplo de chave privada PEM

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

   Copy

2. A partir da linha de comando, gere a chave pública, referenciando a chave privada:

   Supondo que a chave privada esteja criptografada e contida no arquivo chamado rsa_key.p8, use o seguinte comando:

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

   Copy

   Exemplo de chave pública PEM

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

   Copy

3. Copie os arquivos de chaves públicas e privadas em um diretório local para armazenamento. Anote o caminho para os arquivos. A chave privada é armazenada usando o formato PKCS#8 (Public Key Cryptography Standards) e é criptografada usando a frase secreta especificada na etapa anterior; entretanto, o arquivo ainda deve ser protegido contra acesso não autorizado usando o mecanismo de permissão de arquivo fornecido por seu sistema operacional. É responsabilidade dos usuários proteger o arquivo quando ele não está em uso.

4. Faça login no Snowflake. Atribua a chave pública ao usuário do Snowflake usando ALTER USER.

   For example:

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

   Copy

   Nota

   • Somente administradores de segurança (isto é, usuários com a função SECURITYADMIN ou superior) podem alterar um usuário.

   • Exclua o cabeçalho e o rodapé da chave pública na instrução SQL.

   Verifique a impressão digital da chave pública do usuário usando 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

   Nota

   A propriedade RSA_PUBLIC_KEY_2_FP está descrita em Configuração da rotação do par de chaves.

5. Copie e cole a chave privada inteira no campo snowflake.private.key no arquivo de configuração. Salve o arquivo.

Modo distribuído¶

Em uma janela terminal, execute o seguinte comando:

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

Modo autônomo¶

Em uma janela terminal, execute o seguinte comando:

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