Apache Kafka mit DLQ und Metadaten¶

Meldungsformat und Schemaparameter¶

Parameter für die Schema Registry¶

Parameter für DLQ und erweiterte Features¶

Schematization enabled¶

Wenn Schematisierung aktiviert ist, wird der Konnektor:

• Erzeugt einzelne Spalten für jedes Feld in der Meldung

• Enthält eine RECORD_METADATA-Spalte mit Kafka-Metadaten.

• Entwickelt das Tabellenschema automatisch weiter, wenn neue Felder entdeckt werden.

• Vereinfacht verschachtelte JSON/AVRO-Strukturen in separate Spalten.

Beispiel einer Tabellenstruktur:

Schematization disabled¶

Wenn die Schematisierung deaktiviert ist, wird der Konnektor:

• nur zwei Spalten erzeugen: RECORD_CONTENT und RECORD_METADATA

• den gesamten Inhalt der Meldung als OBJECT in RECORD_CONTENT speichern

• keine automatische Schemaentwicklung durchführen

• maximale Flexibilität für die nachgelagerte Verarbeitung bieten

Beispiel einer Tabellenstruktur:

Verwenden Sie die Eigenschaft Schematization Enabled in den Konfigurationseigenschaften des Konnektors, um die Schemaerkennung zu aktivieren oder zu deaktivieren.

Aktivieren der Schema-Entwicklung¶

Wenn der Konnektor die Zieltabelle erstellt, ist die Schemaentwicklung standardmäßig aktiviert.

Wenn Sie die Schemaentwicklung für eine bestehende Tabelle aktivieren oder deaktivieren möchten, verwenden Sie den Befehl ALTER TABLE, um den Parameter ENABLE_SCHEMA_EVOLUTION festzulegen. Sie müssen außerdem eine Rolle verwenden, die über die OWNERSHIP-Berechtigung für die Tabelle verfügt. Weitere Informationen dazu finden Sie unter Tabellenschemaentwicklung.

Wenn jedoch die Schemaentwicklung für eine vorhandene Tabelle deaktiviert ist, versucht der Konnektor, die Zeilen mit nicht übereinstimmenden Schemata an die konfigurierten Dead-Letter-Queues (DLQ) zu senden.

RECORD_METADATA-Struktur¶

Die Spalte RECORD_METADATA enthält wichtige Metadaten zu Kafka-Meldungen:

DLQ-Verhalten¶

• Parse-Fehler – Meldungen mit ungültigem JSON/AVRO-Format werden an die DLQ gesendet.

• Schemaabweichungen – Meldungen, die nicht mit dem erwarteten Schema übereinstimmen, wenn die Schemaentwicklung deaktiviert ist

• Verarbeitungsfehler – Andere Verarbeitungsfehler während der Datenaufnahme

Anforderungen und Einschränkungen¶

Bevor Sie den Openflow Kafka-Konnektor für die Aufnahme von Iceberg-Tabellen konfigurieren, beachten Sie die folgenden Anforderungen und Beschränkungen:

• Sie müssen eine Iceberg-Tabelle erstellen, bevor Sie den Konnektor ausführen.

• Stellen Sie sicher, dass der Benutzer Daten in die erstellten Tabellen eingeben kann.

Konfiguration und Einrichtung¶

Um den Openflow Connector für Kafka für die Aufnahme von Iceberg-Tabellen zu konfigurieren, folgen Sie den Schritten unter Einrichten von Openflow Connector für Kafka mit einigen Abweichungen, die in den folgenden Abschnitten erläutert werden.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    record_metadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "date_added": "2024-10-15"
}

Beispiele für die Erstellung von Iceberg-Tabellen¶

Mit aktivierter Schematisierung:

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    id INT,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

Mit deaktivierter Schematisierung:

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    RECORD_CONTENT OBJECT(
        id INT,
        body_temperature FLOAT,
        name STRING,
        approved_coffee_types ARRAY(STRING),
        animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
        date_added DATE
    )
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';