Working with the Snowflake High Performance connector for Kafka¶

Nomes de tópicos, tabelas e canais¶

Dependendo das definições de configuração, o conector usará nomes diferentes para a tabela de destino. O nome da tabela de destino é sempre derivado do nome do tópico.

Explicando o RECORD_METADATA¶

O conector preenche a estrutura de RECORD_METADATA com os metadados sobre o registro do Kafka. Esses metadados são enviados por meio da fonte de dados Snowpipe Streaming para o Snowflake, onde ficam disponíveis em transformações de canal usando o acessador $1:RECORD_METADATA. A estrutura de RECORD_METADATA está disponível nos modos de canal tanto definido pelo usuário quanto padrão. O conteúdo pode ser salvo na coluna do tipo VARIANT, ou os campos individuais podem ser extraídos e salvos em colunas separadas.

Exemplo de canal com transformações e metadados:

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'))
);

Neste exemplo:

• O canal extrai campos específicos da mensagem do Kafka (order_id, customer_name, order_total)

• Ele também captura campos de metadados (tópico, deslocamento e carimbo de data/hora da ingestão)

• Os valores podem ser convertidos e/ou transformados conforme necessário

Como os campos de metadados são preenchidos¶

O conector preenche automaticamente os campos de metadados com base nas propriedades do registro do Kafka e na configuração do conector. Você pode controlar quais campos de metadados são incluídos usando estes parâmetros de configuração:

• snowflake.metadata.topic (padrão: true): inclui o nome do tópico

• snowflake.metadata.offset.and.partition (padrão: true): inclui deslocamento e partição

• snowflake.metadata.createtime (padrão: true): inclui o carimbo de data/hora do registro do Kafka

• snowflake.metadata.all (padrão: true): inclui todos os metadados disponíveis

Quando snowflake.metadata.all=true (padrão), todos os campos de metadados são preenchidos. A definição de sinalizadores de metadados individuais como false exclui esses campos específicos da estrutura de RECORD_METADATA.

Por padrão, a estrutura de RECORD_METADATA contém as seguintes informações:

A quantidade de metadados gravados na coluna RECORD_METADATA é configurável usando propriedades opcionais de configuração do Kafka.

Os nomes e valores dos campos diferenciam maiúsculas e minúsculas.

Como os registros do Kafka são convertidos antes da ingestão¶

Antes de cada linha ser passada ao Snowpipe Streaming, o conector converte o valor do registro do Kafka Connect em um Map<cadeia de caracteres, objeto>, com chaves que devem corresponder aos nomes das colunas de destino (ou podem ser transformadas dentro de um canal definido pelo usuário). Cadeias de caracteres, matrizes de bytes ou números primitivos devem ser delimitados (por exemplo, usando uma SMT HoistField) para que o conector receba um objeto estruturado. O conversor aplica as seguintes regras:

• Os valores nulos são tratados como marcas de exclusão. Eles são ignorados quando behavior.on.null.values=IGNORE, caso contrário, são ingeridos como objetos JSON vazios.

• Os campos numéricos e boolianos são passados como estão. Valores decimais com precisão maior que 38 são serializados como cadeias de caracteres para permanecer dentro dos limites de NUMBER do Snowflake.

• As cargas úteis byte[] e ByteBuffer são cadeias de caracteres codificadas em Base64, portanto armazene-as em colunas VARIANT ou VARCHAR.

• As matrizes permanecem como matrizes, e os objetos aninhados permanecem como mapas aninhados. Declare as colunas VARIANT quando você recorrer ao canal padrão para descarregar os dados aninhados como estão.

• Mapas com chaves que não são cadeia de caracteres são emitidos como matrizes de pares [key, value] porque os nomes das colunas do Snowflake devem ser texto.

• Os cabeçalhos e as chaves dos registros são copiados para RECORD_METADATA sempre que os sinalizadores de metadados relevantes estiverem habilitados.

Se você precisar que todo o corpo da mensagem seja preservado como uma única coluna, delimite-o em um novo campo de nível superior usando SMTs. Consulte Coluna RECORD_CONTENT legada para ver o padrão de transformação.

Modo de canal definido pelo usuário¶

Neste modo, você tem controle completo sobre a transformação de dados e o mapeamento de colunas.

Quando usar este modo:

• Você precisa de nomes de colunas personalizados que sejam diferentes dos nomes de campo JSON

• Você precisa aplicar transformações de dados (conversão de tipo, mascaramento, filtragem)

• Você quer controle completo sobre como os dados são mapeados para colunas

Modo de canal padrão¶

Neste modo, o conector usa um canal padrão chamado {tableName}-STREAMING e mapeia os campos de registro do Kafka para colunas de tabelas correspondentes por nome (sem distinção entre maiúsculas e minúsculas).

Quando usar este modo:

• Os nomes da sua chave de registro do Kafka correspondem aos nomes das colunas desejadas

• Você não precisa de transformações de dados personalizadas

• You want a simple configuration

Mapeando chaves de registro do Kafka para colunas de tabela com modo de canal padrão

Ao usar o modo de canal padrão, o conector usa o canal padrão chamado {tableName}-STREAMING e mapeia as chaves de primeiro nível do conteúdo diretamente para as colunas da tabela usando uma correspondência sem distinção entre maiúsculas e minúsculas.

Usando o modo de canal padrão – exemplo¶

{
  "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
);

Usando o modo de canal definido pelo usuário – exemplos¶

Este exemplo mostra como configurar e usar canais definidos pelo usuário com transformações de dados personalizadas.

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'))
);

Mapeamento explícito de tópico para tabela¶

Quando você configura o parâmetro snowflake.topic2table.map, o conector opera no modo de mapeamento explícito. Esse modo permite que você:

• Mapeie vários tópicos do Kafka para uma única tabela do Snowflake;

• Use nomes de tabelas personalizados diferentes dos nomes dos tópicos;

• Aplique padrões regex para corresponder a vários tópicos.

Formato de configuração:

O parâmetro snowflake.topic2table.map aceita uma lista separada por vírgulas de mapeamentos de tópico para tabela no formato:

topic1:table1,topic2:table2,topic3:table3

Exemplos de configurações:

Mapeamento direto do tópico

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

Correspondência de padrão regex

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

Esta configuração mapeia todos os tópicos que terminam com _cat (como orange_cat, calico_cat) para a tabela CAT_TABLE, e todos os tópicos que terminam com _dog para a tabela DOG_TABLE.

Muitos tópicos para uma tabela

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

Esta configuração mapeia tanto topic1 quanto topic2 para shared_table, enquanto topic3 é mapeado para other_table.

Coluna RECORD_CONTENT legada¶

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'))
);

Se você precisa de todo o conteúdo da mensagem do Kafka salvo em uma única coluna, ou de um manipulador para todo o conteúdo da mensagem do Kafka em uma transformação de PIPE, pode usar a seguinte transformação SMT que delimita todo o conteúdo da mensagem do Kafka no campo personalizado desejado:

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

Essa transformação delimita todo o conteúdo da mensagem do Kafka em um campo personalizado chamado your_top_level_field_name. Em seguida, você pode acessar todo o conteúdo da mensagem do Kafka usando o acessador $1:your_top_level_field_name em sua transformação de PIPE.

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

Uma outra opção se você quiser salvar todos os metadados e o conteúdo em uma única tabela usando o canal padrão é não criar um canal personalizado; em vez disso, crie apenas uma tabela com duas colunas: RECORD_CONTENT e your_top_level_field_name.

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

Para ler mais sobre a transformação HoistField$Value, consulte a Documentação do Kafka.

Tratamento de erros de canais de streaming e filas de mensagens mortas¶

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

Concessão de uso em um volume externo¶

Para conceder o privilégio USAGE no volume externo associado à tabela Apache Iceberg™ à sua função para o conector Kafka, execute a seguinte instrução:

Por exemplo, se a tabela Iceberg usar o volume externo kafka_external_volume e o conector usar a função kafka_connector_role, execute a seguinte instrução:

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

Criar uma tabela Apache Iceberg™ para ingestão¶

O conector não cria tabelas Iceberg automaticamente e não oferece suporte à evolução de esquema. Antes de executar o conector, você deve criar uma tabela Iceberg manualmente.

Ao criar uma tabela Iceberg, é possível usar tipos de dados Iceberg (incluindo VARIANT) ou tipos compatíveis com o Snowflake.

Por exemplo, considere a seguinte mensagem:

{
    "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"
}

Para criar uma tabela Iceberg para a mensagem de exemplo, use uma das seguintes instruções:

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