Kafkaコネクタのトラブルシューティング¶

生バイト¶

ファイル名 <コネクタ名>/<テーブル名>/<パーティション>/offset_(<キー>/<値>_)<タイムスタンプ>.gz には、生バイトからソースファイル形式に変換されなかった記録の正確なオフセットが含まれます。問題を解決するには、記録を新しい記録としてKafkaコネクタに再送信します。

ソースファイル形式（Avro、 JSON、またはProtobuf）¶

Snowpipeが、Kafkaトピック用に作成された内部ステージのファイルからデータをロードできなかった場合、ソースファイル形式にあるターゲットテーブルのために、Kafkaコネクタはファイルをステージに移動します。

ファイルのセットに複数の問題がある場合、 COPY_HISTORY 出力の FIRST_ERROR_MESSAGE 列は最初に発生したエラーのみを示します。ファイル内のすべてのエラーを表示するには、テーブルステージからファイルを取得し、それらを名前付きステージにアップロードしてから、 VALIDATION_MODE コピーオプションを RETURN_ALL_ERRORS に設定して COPY INTO <テーブル> ステートメントを実行する必要があります。VALIDATION_MODE コピーオプションは、ロードされるデータを検証し、指定された検証オプションに基づいて結果を返すように COPY ステートメントに指示します。このコピーオプションが指定されている場合、データはロードされません。ステートメントで、Kafkaコネクタを使用してロードしようとしたファイルのセットを参照します。

データファイルの問題が解決したら、1つ以上の COPY ステートメントを使用してデータを手動でロードできます。

次の例は、 mydb.public データベースとスキーマの mytable テーブルのテーブルステージにあるデータファイルを参照します。

テーブルステージでデータファイルを検証し、エラーを解決するには、

1. LIST を使用して、テーブルステージにあるすべてのファイルを一覧表示します。

   例:

   LIST @mydb.public.%mytable;
   

   Copy

   このセクションの例では、 JSON がデータファイルのソース形式であると想定しています。

2. GET を使用して、Kafkaコネクタによって作成されたファイルをローカルマシンにダウンロードします。

   たとえば、ローカルマシンの data という名前のディレクトリにファイルをダウンロードします。

   Linuxまたは macOS:

   GET @mydb.public.%mytable file:///data/;
   

   Copy

   Microsoft Windows:

   GET @mydb.public.%mytable file://C:\data\;
   

   Copy

3. CREATE STAGE を使用して、ソースKafkaファイルと同じ形式のデータファイルを保存する名前付き内部ステージを作成します。

   たとえば、 JSON ファイルを保存する kafka_json という名前の内部ステージを作成します。

   CREATE STAGE kafka_json FILE_FORMAT = (TYPE = JSON);
   

   Copy

4. PUT を使用して、テーブルステージからダウンロードしたファイルをアップロードします。

   たとえば、ローカルマシンの data ディレクトリにダウンロードしたファイルをアップロードします。

   Linuxまたは macOS:

   PUT file:///data/ @mydb.public.kafka_json;
   

   Copy

   Microsoft Windows:

   PUT file://C:\data\ @mydb.public.kafka_json;
   

   Copy

5. テスト目的で、2つのバリアント列を持つ仮テーブルを作成します。このテーブルは、ステージングされたデータファイルを検証する目的のみに使用されます。データはテーブルにロードされません。現在のユーザーセッションが終了すると、テーブルは自動的にドロップされます。

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. COPY INTO *テーブル* ... VALIDATION_MODE = 'RETURN_ALL_ERRORS' ステートメントを実行して、データファイルで発生したすべてのエラーを取得します。このステートメントは、指定されたステージでファイルを検証します。データはテーブルにロードされません。

   COPY INTO mydb.public.t1
     FROM @mydb.public.kafka_json
     FILE_FORMAT = (TYPE = JSON)
     VALIDATION_MODE = 'RETURN_ALL_ERRORS';
   

   Copy

7. ローカルマシンのデータファイルで報告されたすべてのエラーを修正します。

8. PUT を使用して、固定ファイルをテーブルステージまたは指定された内部ステージのいずれかにアップロードします。

   次の例では、ファイルをテーブルステージにアップロードし、既存のファイルを上書きします。

   Linuxまたは macOS:

   PUT file:///tmp/myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

   Windows:

   PUT file://C:\temp\myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

9. VALIDATION_MODE オプション なし で COPY INTO テーブル を使用し、データをターゲットテーブルにロードします。

   オプションで PURGE = TRUE コピーオプションを使用して、データが正常にロードされたらステージからデータファイルを削除するか、 REMOVE を使用して、テーブルステージから手動でファイルを削除します。

   COPY INTO mydb.public.mytable(RECORD_METADATA, RECORD_CONTENT)
     FROM (SELECT $1:meta, $1:content FROM @mydb.public.%mytable)
     FILE_FORMAT = (TYPE = 'JSON')
     PURGE = TRUE;
   

   Copy