Working with the Snowflake High Performance connector for Kafka¶

Topic-Namen, Tabellennamen und Pipe-Namen¶

Je nach Konfigurationseinstellung verwendet der Konnektor unterschiedliche Namen für die Zieltabelle. Der Name der Zieltabelle wird immer vom Namen des Topics abgeleitet.

Erläuterungen zu RECORD_METADATA¶

Der Konnektor füllt die RECORD_METADATA-Struktur mit Metadaten zum Kafka-Datensatz auf. Diese Metadaten werden über die Snowpipe Streaming-Datenquelle an Snowflake gesendet, wo sie mithilfe des $1:RECORD_METADATA-Accessors in Pipe-Transformationen verfügbar gemacht werden. Die RECORD_METADATA-Struktur ist sowohl in benutzerdefinierten Pipe- als auch in Standard-Pipe-Modi verfügbar. Ihr Inhalt kann in der Spalte des Typs VARIANT gespeichert werden, oder einzelne Felder können extrahiert und in separaten Spalten gespeichert werden.

Beispiel für Pipe mit Transformationen und Metadaten:

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

In diesem Beispiel:

• Die Pipe extrahiert bestimmte Felder aus der Kafka-Nachricht (order_id, customer_name, order_total)

• Es werden auch Metadatenfelder erfasst (Topics, Offset und Zeitstempel der Datenaufnahme).

• Die Werte können nach Bedarf umgewandelt und/oder transformiert werden.

Wie die Metadatenfelder ausgefüllt werden¶

Der Konnektor füllt die Metadatenfelder automatisch auf der Grundlage der Eigenschaften des Kafka-Datensatzes und der Konfiguration des Konnektors aus. Mit diesen Konfigurationsparametern können Sie steuern, welche Metadatenfelder enthalten sind:

• snowflake.metadata.topic (Standard: true) – Beinhaltet den Themennamen

• snowflake.metadata.offset.and.partition (Standard: „true“) – Beinhaltet Offset und Partition

• snowflake.metadata.createtime (Standard: true) – Beinhaltet den Zeitstempel des Kafka-Datensatzes

• snowflake.metadata.all (Standard: true) – Beinhaltet alle verfügbaren Metadaten

Wann snowflake.metadata.all=true (Standard) ist, werden alle Metadatenfelder ausgefüllt. Das Festlegen einzelner Metadaten-Flags auf false schließt diese spezifischen Felder aus der RECORD_METADATA-Struktur aus.

Die Spalte RECORD_METADATA enthält standardmäßig die folgenden Informationen:

Die Menge der in der Spalte RECORD_METADATA aufgezeichneten Metadaten kann mithilfe der optionalen Kafka-Konfigurationseigenschaften konfiguriert werden.

Bei den Feldnamen und Werten muss die Groß- und Kleinschreibung beachtet werden.

Wie Kafka-Datensätze vor der Erfassung konvertiert werden¶

Bevor jede Zeile an Snowpipe Streaming übergeben wird, konvertiert der Konnektor den Wert des Kafka Connect-Datensatzes in eine Map<Zeichenfolge, Objekt>, deren Schlüssel mit den Namen der Zielspalten übereinstimmen müssen (oder in einer benutzerdefinierten Pipe umgewandelt werden können). Primitive Zeichenfolgen, Byte-Arrays oder Zahlen müssen umschlossen werden (z. B. durch Verwendung von HoistField SMT), sodass der Konnektor ein strukturiertes Objekt erhält. Der Konverter wendet die folgenden Regeln an:

• Nullwerte werden als Tombstones behandelt. Sie werden übersprungen, wenn behavior.on.null.values=IGNORE ist, oder ansonsten als leere JSON-Objekte aufgenommen.

• Numerische und boolesche Felder werden unverändert übergeben. Dezimalwerte, deren Genauigkeit größer als 38 ist, werden als Zeichenfolgen serialisiert, um innerhalb der NUMBER-Grenzen von Snowflake zu bleiben.

• byte[]- und ByteBuffer-Payloads sind Base64-kodierte Zeichenfolgen; speichern Sie sie daher in den Spalten VARIANT oder VARCHAR.

• Arrays bleiben Arrays, und verschachtelte Objekte bleiben verschachtelte Zuordnungen. Deklarieren Sie VARIANT-Spalten, wenn Sie sich auf die Standard-Pipe verlassen, um verschachtelte Daten so zu übertragen, wie sie sind.

• Zuordnungen mit Nicht-Zeichenfolgen-Schlüsseln werden als Arrays von [key, value]-Paaren ausgegeben, da Snowflake-Spaltennamen Text sein müssen.

• Datensatzheader und -schlüssel werden immer dann in RECORD_METADATA kopiert, wenn die entsprechenden Metadaten-Flags aktiviert sind.

Wenn Sie möchten, dass der gesamte Nachrichtentext als eine einzelne Spalte beibehalten wird, schließen Sie ihn mit SMTs in ein neues Feld der obersten Ebene ein. Informationen zum Transformationsmuster finden Sie unter Ältere RECORD_CONTENT-Spalte.

Benutzerdefinierter Pipe-Modus¶

In diesem Modus haben Sie die volle Kontrolle über die Datentransformation und die Spaltenzuordnung.

Wann dieser Modus verwendet werden sollte:

• Sie benötigen benutzerdefinierte Spaltennamen, die sich von den JSON-Feldnamen unterscheiden.

• Sie müssen Datentransformationen anwenden (Typumwandlung, Maskierung, Filterung)

• Sie möchten die volle Kontrolle darüber haben, wie Daten den Spalten zugeordnet werden

Standard-Pipe-Modus¶

In diesem Modus verwendet der Konnektor eine Standard-Pipe namens {tableName}-STREAMING und ordnet Kafka-Datensatzfelder Tabellenspalten zu, die dem Namen nach übereinstimmen (Groß-/Kleinschreibung wird nicht berücksichtigt).

Wann dieser Modus verwendet werden sollte:

• Die Namen der Kafka-Datensatzschlüssel entsprechen den gewünschten Spaltennamen

• Sie benötigen keine benutzerdefinierten Datentransformationen

• You want a simple configuration

Zuordnung von Kafka-Datensatzschlüsseln zu Tabellenspalten mit Standard-Pipe-Modus

Wenn Sie den Standard-Pipe-Modus verwenden, verwendet der Konnektor eine Standard-Pipe namens {tableName}-STREAMING und ordnet die Schlüssel der ersten Ebene des Inhalts direkt den Tabellenspalten zu, wobei die Groß-/Kleinschreibung nicht berücksichtigt wird.

Verwendung des Standard-Pipe-Modus – Beispiel¶

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

Verwenden des benutzerdefinierten Pipe-Modus – Beispiele¶

Dieses Beispiel zeigt, wie Sie benutzerdefinierte Pipes mit benutzerdefinierten Datentransformationen konfigurieren und verwenden.

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

Explizite Zuordnung von Topics zu Tabellen¶

Wenn Sie den Parameter snowflake.topic2table.map konfigurieren, arbeitet der Konnektor im expliziten Zuordnungsmodus. Dieser Modus bietet Folgendes:

• Zuordnen mehrerer Kafka-Topics zu einer einzigen Snowflake-Tabelle

• Verwenden von benutzerdefinierten Tabellennamen, die sich von Topic-Namen unterscheiden

• Anwenden von Regex-Mustern, um mehrere Topics zuzuordnen

Konfigurationsformat:

Der snowflake.topic2table.map-Parameter akzeptiert eine durch Kommas getrennte Liste von Topics-zu-Topics-Zuordnungen im folgenden Format:

topic1:table1,topic2:table2,topic3:table3

Beispielkonfigurationen:

Direkte Topic-Zuordnung

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

Regex-Mustererkennung

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

Dieses Konfiguration ordnet alle Topics, die mit _cat enden (wie z. B. orange_cat, calico_cat) der Tabelle CAT_TABLE zu, und alle Topics, die mit _dog enden, werden der Tabelle DOG_TABLE zugeordnet.

Viele Topics zu einer Tabelle

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

Diese Konfiguration ordnet sowohl topic1 als auch topic2 der shared_table zu, während topic3 der other_table zugeordnet wird.

Ältere RECORD_CONTENT-Spalte¶

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

Wenn Sie den gesamten Inhalt einer Kafka-Nachricht in einer einzigen Spalte speichern möchten, oder wenn Sie ein Handle für den gesamten Inhalt der Kafka-Nachricht in einer PIPE-Transformation benötigen, können Sie die folgende SMT-Transformation verwenden, die den gesamten Inhalt der Kafka-Nachricht in das gewünschte benutzerdefinierte Feld einschließt:

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

Durch diese Transformation wird der gesamte Inhalt der Kafka-Nachricht in ein benutzerdefiniertes Feld mit dem Namen your_top_level_field_name eingeschlossen. Sie können dann mithilfe der des $1:your_top_level_field_name-Accessors in Ihrer PIPE-Transformation auf den gesamten Inhalt der Kafka-Nachricht zugreifen.

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

Wenn Sie alternativ die gesamten Metadaten und den Inhalt in einer einzigen Tabelle mit der Standard-Pipe speichern möchten, erstellen Sie keine benutzerdefinierte Pipe. Erstellen Sie stattdessen eine Tabelle mit nur zwei Spalten: RECORD_CONTENT und your_top_level_field_name.

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

Weitere Informationen zur HoistField$Value-Transformation finden Sie in der Kafka-Dokumentation.

Behandlung von Streaming-Kanal-Fehlern und Dead-Letter-Queues¶

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

Verwendung auf einem externen Volume gewähren¶

Um die USAGE-Berechtigung für das externe Volume, das mit Ihrer Apache Iceberg™-Tabelle verknüpft ist, an Ihre Rolle für den Kafka-Konnektor zu erteilen, führen Sie die folgende Anweisung aus:

Wenn Ihre Iceberg-Tabelle beispielsweise das externe Volume kafka_external_volume verwendet und der Konnektor die Rolle kafka_connector_role benutzt, führen Sie die folgende Anweisung aus:

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

Erstellen einer Apache Iceberg™-Tabelle für die Datenaufnahme¶

Der Konnektor erstellt Iceberg-Tabellen nicht automatisch und unterstützt keine Schemaentwicklung. Bevor Sie den Konnektor ausführen, müssen Sie manuell eine Iceberg-Tabelle erstellen.

Wenn Sie eine Iceberg-Tabelle erstellen, können Sie Iceberg-Datentypen (wie VARIANT) oder kompatible Snowflake-Typen verwenden.

Betrachten Sie zum Beispiel die folgende Meldung:

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

Um eine Iceberg-Tabelle für die Beispielmeldung zu erstellen, verwenden Sie eine der folgenden Anweisungen:

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