DLQ とメタデータを使用したApache Kafka¶

主要な機能¶

DLQ とメタデータコネクタを使用したApache Kafkaは包括的な機能を提供します。

• デッドレターキュー（DLQ） による失敗メッセージ処理のサポート

• Kafkaメッセージのメタデータがある RECORD_METADATA 列

• 構成可能なスキーマ化 - スキーマ検出の有効化または無効化

• スキーマが進化する Icebergテーブルのサポート

• 複数のメッセージ形式 - JSON および AVRO のサポート

• AVRO メッセージ用の スキーマレジストリ統合

• 高度なパターンがある トピックからテーブルへのマッピング

• SASL 認証 のサポート

固有パラメーター¶

Openflow Connector for Kafkaを設定する で説明されている一般的なパラメーターに加えて、このコネクタには高度な機能のための追加のパラメーターコンテキストが含まれます。

スキーマ化の動作¶

コネクタの動作は、 Schematization Enabled パラメーターに基づいて変更されます。

スキーマの検出と進化¶

コネクタはスキーマの検出と進化をサポートしています。Snowflakeのテーブルの構造は、コネクタによってロードされた新しいデータの構造をサポートするように定義し、自動的に進化させることができます。

スキーマ検出と進化を行わない場合、コネクタによってロードされたSnowflakeテーブルは、 RECORD_CONTENT と RECORD_METADATA の2つの OBJECT 列のみで構成されます。

スキーマ検出と進化がある場合、Snowflakeはストリーミングデータのスキーマを検出し、ユーザー定義スキーマに自動的に一致するテーブルにデータをロードします。Snowflakeでは、新しい列を追加したり、新しいデータファイルで欠落している列から NOT NULL 制約をドロップしたりすることもできます。

コネクタによるスキーマ検出は、提供されたスキーマレジストリの有無にかかわらずサポートされています。スキーマレジストリ（Avro）を使用する場合、列はスキーマレジストリで定義されたデータ型で作成されます。スキーマレジストリ（JSON）がない場合、データ型は提供されるデータに基づいて推測されます。

それ以上のスキーマ化ではJSONARRAYはサポートされていません。

デッドレターキュー（DLQ）¶

DLQ 機能は、正常に処理できないメッセージを処理します。

Icebergテーブルのサポート¶

Openflow Connector for Kafkaは、 Iceberg Enabled が true に設定されている場合、Snowflake管理の Apache Iceberg™ テーブル にデータを取り込むことができます。

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

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

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

ユースケース¶

このコネクタは次のようなケースに最適です。

• DLQ が必要な 本番環境

• Kafkaメタデータが重要な データ系統と監査

• スキーマ進化要件のある 複雑なメッセージ処理

• Icebergテーブル統合

メタデータや DLQ 機能のないよりシンプルな取り込みが必要な場合は、代わりに JSON/AVRO データ形式用Apache Kafka コネクタを検討してください。