Kafka Connector 故障排除¶

原始字节¶

文件名 <connector_name>/<table_name>/<partition>/offset_(<key>/<value>_)<timestamp>.gz 包含未从原始字节转换为源文件格式的记录的确切偏移。要解决问题，请将该记录作为新记录重新发送到 Kafka Connector。

源文件格式（Avro、JSON 或 Protobuf）¶

如果 Snowpipe 无法从为 Kafka 主题创建的内部暂存区中的文件加载数据，Kafka Connector 会将文件以源文件格式移动到目标表的暂存区。

如果一组文件存在多个问题，则 COPY_HISTORY 输出中的 FIRST_ERROR_MESSAGE 列仅指明遇到的第一个错误。要查看文件中的所有错误，必须从表暂存区检索文件，将其上传到已命名暂存区，然后执行将 VALIDATION_MODE 复制选项设置为 RETURN_ALL_ERRORS 的 COPY INTO <table> 语句。VALIDATION_MODE 复制选项指示 COPY 语句验证要加载的数据并根据指定的验证选项返回结果。指定此复制选项时，不会加载任何数据。在语句中，引用您尝试使用 Kafka Connector 加载的一组文件。

当数据文件的任何问题得到解决后，您可以使用一个或多个 COPY 语句手动加载数据。

以下示例引用了位于 mydb.public 数据库和架构中 mytable 表的表暂存区的数据文件。

要在表暂存区验证数据文件并解决错误，请执行以下操作：

1. 使用 LIST 列出位于表暂存区的所有文件。

   例如：

   LIST @mydb.public.%mytable;
   

   Copy

   本节中的示例假定 JSON 是数据文件的源格式。

2. 使用 GET 将 Kafka Connector 创建的文件下载至本地计算机。

   例如，将文件下载到本地计算机中名为 data 的目录中：

   Linux 或 macOS:

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

   Copy

   Microsoft Windows:

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

   Copy

3. 使用 CREATE STAGE 创建指定内部暂存区，用于存储与源 Kafka 文件格式相同的数据文件。

   例如，创建一个名为 kafka_json 的内部暂存区来存储 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. 创建一个包含两列变体的临时表进行测试。该表仅用于验证暂存数据文件。表中未加载任何数据。当前用户会话结束时，该表将自动删除：

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. 执行 COPY INTO *table* ...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 table 将数据加载到目标表中。

   成功加载数据后，您可以选择使用 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