Funktionsweise von 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

• Sie wünschen eine einfache Konfiguration

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¶

Wenn in älteren Versionen des Konnektors die Schematisierungsfunktion deaktiviert war, erstellte der Konnektor eine Zieltabelle mit zwei Spalten: RECORD_CONTENT und RECORD_METADATA. Die RECORD_CONTENT-Spalte enthielt den gesamten Kafka-Nachrichtinhalt in einer Spalte vom Typ VARIANT. Die RECORD_METADATA-Spalte wird weiterhin unterstützt, aber die RECORD_CONTENT-Spalte wird nicht mehr vom Konnektor erstellt. Die gleiche Funktionalität kann mit SMT-Transformationen erreicht werden (siehe Beispiele weiter unten in diesem Abschnitt). Der RECORD_CONTENT-Schlüssel ist auch nicht mehr in PIPE-Transformationen verfügbar. Diese PIPE-Definition funktioniert beispielsweise nicht standardmäßig:

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¶

In Version 4.0.0-rc4 prüft der Konnektor den Status des Snowpipe Streaming-Kanals, bevor Offsets übertragen werden. Wenn Snowflake abgelehnte Zeilen meldet (rowsErrorCount > 0), the connector now raises a fatal error (ERROR_5030) when errors.tolerance=none so that data issues cannot go unnoticed. To allow ingestion to continue while you triage bad rows, set errors.tolerance=all

errors.tolerance=all

Einschränkungen der Fehlertoleranz mit dem Konnektor¶

Kafka-Themen können mit einer Begrenzung des Speicherplatzes oder der Aufbewahrungsdauer konfiguriert werden.

• Wenn das System länger als die Aufbewahrungsdauer offline ist, werden abgelaufene Datensätze nicht geladen. Auf ähnliche Weise werden einige Nachrichten nicht zugestellt, wenn das Speicherplatzlimit von Kafka überschritten wurde.

• Wenn Nachrichten im Kafka-Topic gelöscht oder aktualisiert werden, spiegeln sich diese Änderungen ggf. nicht in der Snowflake-Tabelle wider.