Snowflake High Performance connector for Kafka の操作¶

トピック名、テーブル名、パイプ名¶

構成設定に応じて、コネクタは宛先テーブルに異なる名前を使用します。宛先テーブル名は、常にトピック名から派生します。

RECORD_METADATA を理解する¶

コネクタは、Kafka記録に関するメタデータを含む RECORD_METADATA 構造を入力します。このメタデータは、Snowpipe Streamingデータソースを通じてSnowflakeに送信され、Snowflakeでは $1:RECORD_METADATA アクセサーを使用してパイプ変換で利用可能になります。RECORD_METADATA 構造は、ユーザー定義パイプとデフォルトパイプモードの両方で使用できます。その内容は、 VARIANT 型の列に保存できます。または個々のフィールドを抽出して、個別の列に保存できます。

変換とメタデータを含むパイプの例:

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

この例では:

• パイプは、Kafkaメッセージから特定のフィールド（order_id、customer_name、order_total）を抽出します。

• また、メタデータフィールド（トピック、オフセット、インジェスションタイムスタンプ）もキャプチャします。

• 値は、必要に応じてキャストおよび/または変換できます

メタデータフィールドの入力方法¶

コネクタは、Kafka記録プロパティとコネクタ構成に基づいてメタデータフィールドを自動的に入力します。これらの構成パラメーターを使用して、どのメタデータフィールドを含めるかを制御できます。

• snowflake.metadata.topic （デフォルト: true） - トピック名を含む

• snowflake.metadata.offset.and.partition （デフォルト: true） - オフセットとパーティションを含む

• snowflake.metadata.createtime （デフォルト: true） - Kafka記録のタイムスタンプを含む

• snowflake.metadata.all （デフォルト: true） - 利用可能なすべてのメタデータを含む

snowflake.metadata.all=true （デフォルト）の場合、すべてのメタデータフィールドが入力されます。個々のメタデータフラグを false に設定すると、 RECORD_METADATA 構造から特定のフィールドを除外します。

RECORD_METADATA 構造には、デフォルトで次の情報が含まれています。

RECORD_METADATA 列に記録されるメタデータの量は、オプションのKafka構成プロパティを使用して構成できます。

フィールド名と値は大文字と小文字が区別されます。

インジェスション前のKafka記録の変換方法¶

各行がSnowpipe Streamingに渡される前に、コネクタはKafka Connectの記録値を Map<文字列、オブジェクト> に変換します。このキーはターゲット列名と一致する必要があります（または、ユーザー定義のパイプ内で変換できます）。プリミティブ文字列、バイト配列、または数値は（たとえば HoistField SMT を使用して）ラップする必要があります。それにより、コネクタは構造化オブジェクトを受け取れるようになります。コンバーターは次のルールを適用します。

• Null値はtombstoneとして扱われます。behavior.on.null.values=IGNORE または、空の JSON オブジェクトとして取り込まれる場合はスキップされます。

• 数値およびブールフィールドはそのまま渡されます。精度が38より大きい10進値は、Snowflakeの NUMBER の制限内にとどまる文字列としてシリアル化されます。

• byte[] および ByteBuffer ペイロードはBase64でエンコードされた文字列であるため、 VARIANT または VARCHAR 列に保存します。

• 配列は配列のままであり、ネストされたオブジェクトはネストされたマップのままです。デフォルトのパイプに依存してネストされたデータをそのままにする場合、 VARIANT 列を宣言します。

• 非文字列キーを持つマップは、 [key, value] ペアの配列として出力されます。これは、Snowflakeの列名はテキストである必要があるためです。

• 関連するメタデータフラグが有効になっている場合は常に、記録のヘッダーとキーは RECORD_METADATA にコピーされます。

メッセージ本文全体を単一の列として保持する必要がある場合は、 SMTs を使用して新しいトップレベルフィールドにラップします。 変換パターンについては レガシー RECORD_CONTENT 列 をご参照ください。

ユーザー定義のパイプモード¶

このモードでは、データ変換と列マッピングをフルコントロールできます。

このモードを使用する場合:

• JSON フィールド名とは異なるカスタム列名が必要です。

• データ変換（型キャスト、マスキング、フィルタリング）を適用する必要があります

• データが列にどのようにマッピングされるかをフルコントロールしたい場合

デフォルトのパイプモード¶

このモードでは、コネクタは {tableName}-STREAMING という名前のデフォルトのパイプを使用し、kafkaレコードフィールドを名前（大文字と小文字の区別なし）で一致するテーブル列にマップします。

このモードを使用する場合:

• kafkaレコードのキー名は希望する列名と一致します

• カスタムデータ変換は必要ありません

• 簡単な構成が必要です

デフォルトのパイプモードを使用したテーブル列へのkafka記録キーのマッピング

デフォルトパイプモードを使用すると、コネクタは {tableName}-STREAMING という名前のデフォルトパイプを使用し、大文字と小文字を区別しない一致を使用して、コンテンツの第1レベルキーをテーブル列に直接マッピングします。

デフォルトパイプモードの使用 - 例¶

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

ユーザー定義パイプモードの使用 - 例¶

この例では、カスタムデータ変換でユーザー定義のパイプを構成して使用する方法を示します。

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

明示的なトピックからテーブルへのマッピング¶

snowflake.topic2table.map パラメーターを構成するときは、コネクタは明示的マッピングモードで動作します。このモードでは、次を実行できます。

• 複数のKafkaトピックを単一のSnowflakeテーブルにマッピングする

• トピック名とは異なるカスタムテーブル名を使用する

• 複数のトピックに正規表現パターンを適用する

構成形式:

snowflake.topic2table.map パラメーターは、トピックからテーブルへのマッピングのコンマ区切りリストを以下の形式で受け入れます。

topic1:table1,topic2:table2,topic3:table3

構成例:

直接トピックマッピング

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

正規表現パターンマッチング

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

この構成は、 _cat （ orange_cat 、 calico_cat など）で終わるすべてのトピックを CAT_TABLE テーブルに、 _dog で終わるすべてのトピックを DOG_TABLE テーブルにマップします。

1つのテーブルへの多数のトピック

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

この構成は topic1 および topic2 の両方を shared_table にマップする一方で、 topic3 は other_table にマップします。

レガシー RECORD_CONTENT 列¶

コネクタの以前のバージョン（3.x以前）でスキーマ化機能が無効になっていた場合、コネクタはRECORD_CONTENTおよびRECORD_METADATAの2つの列を持つ宛先テーブルを作成しました。RECORD_CONTENT列には、VARIANT型の列にKafkaメッセージのコンテンツ全体が含まれていました。RECORD_METADATA列は引き続きサポートされていますが、RECORD_CONTENT列はコネクタによって作成されなくなりました。同じ機能は SMT 変換を使用して実現できます（このセクションの後半の例をご参照ください）。RECORD_CONTENT キーも PIPE 変換では利用できなくなりました。たとえば、この PIPE 定義はデフォルトでは機能しません。

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

Kafkaメッセージのコンテンツ全体を単一の列に保存する必要がある場合や、 PIPE 変換でKafkaメッセージのコンテンツ全体の処理が必要な場合、Kafkaメッセージのコンテンツ全体を希望のカスタムフィールドにラップする、次の SMT 変換を使用できます。

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

この変換は、Kafkaメッセージのコンテンツ全体を your_top_level_field_name という名前のカスタムフィールドにラップします。その後は、 PIPE 変換の $1:your_top_level_field_name アクセサーを使用して、Kafkaメッセージのコンテンツ全体にアクセスできます。

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

または、デフォルトのパイプを使用してメタデータとコンテンツの両方を1つのテーブルに保存する場合は、カスタムパイプを作成しないでください。代わりに、 RECORD_CONTENT および your_top_level_field_name の2つの列を持つテーブルのみを作成します。

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

HoistField$Valueの変換の詳細を読むには、 Kafkaドキュメント を参照してください。

ストリーミングチャネルエラーと配信不能キューの処理¶

コネクタはKafkaでオフセットをコミットする前にSnowpipe Streamingチャネルのステータスを検査します。コネクタが、コネクタの起動以降にチャネルの rowsErrorCount プロパティの増加を検出した場合、 errors.tolerance=none のときに致命的なエラー（ ERROR_5030 ）が発生し、データの問題が認識されないことを防ぎます。不正な行をトリガーしている間にインジェスチョンが継続されるようにするには、 errors.tolerance=all を設定します。

errors.tolerance=all

外部ボリュームの使用許可¶

Kafkaコネクタのロールに Apache Iceberg™ テーブルに関連付けられた外部ボリュームに対するUSAGE権限を付与するには、次のステートメントを実行します。

たとえば、Icebergテーブルが kafka_external_volume 外部ボリュームを使用し、コネクタがロール kafka_connector_role を使用する場合、次のステートメントを実行します：

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

インジェスチョン用の Apache Iceberg™ テーブルを作成する¶

コネクタは自動的にIcebergテーブルを作成せず、スキーマの進化もサポートしません。コネクタを実行する前に、Icebergテーブルを手動で作成する必要があります。

Icebergテーブルを作成する場合、Icebergデータタイプ（VARIANTなど）または 互換性のあるSnowflakeデータタイプ を使用できます。

例えば、次のようなメッセージを考えてみましょう：

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

例のメッセージ用にIcebergテーブルを作成するには、以下のステートメントの1つを使用します:

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