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 構造には、デフォルトで次の情報が含まれています。

The amount of metadata recorded in the RECORD_METADATA column is configurable using optional Kafka configuration properties.

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

インジェスション前の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 列¶

コネクタの古いバージョンでスキーマ化機能が無効になっていた場合、コネクタは 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ドキュメント を参照してください。

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

バージョン 4.0.0-rc4 では、コネクタはオフセットをコミットする前にSnowpipe Streamingチャネルのステータスを検査します。Snowflakeが拒否された行をレポートする場合（ 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

コネクタのフォールトトレランスの制限¶

Kafkaトピックは、ストレージスペースまたは保持時間の制限で構成できます。

• 保持時間を超えてシステムがオフラインしている場合、期限切れの記録はロードされません。同様に、Kafkaのストレージスペースの制限を超えると、一部のメッセージは配信されません。

• Kafkaトピックのメッセージが削除された場合、これらの変更はSnowflakeテーブルに反映されません。