如何使用 ON_ERROR=CONTINUE 在 Snowpipe 中有效解析 json

我正在设置一个 Snowpipe 将数据从 s3 存储桶加载到雪花模式。 S3 包含 NDJOSN 格式的文件。一个文件可以包含多条记录，我想处理所有这些记录。哪怕打破一项记录。

为此，我需要添加on_error='continue'管道创建和使用 csv 文件格式的选项，如官方雪花文档中所述here https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-table.html#usage-notes.

这样我就可以收到需要解析以访问数据的 JSON 原始字符串。由于 Snowpipe 不支持嵌套选择，唯一的方法就是单独解析每一列。

生成此副本声明：

copy into MY_TABLE 
from (select parse_json($1):id, parse_json($1):name, parse_json($1):status 
from @MY_STAGE_CONNECTED_TO_S3)  
on_error = 'continue'

这段代码需要对每一行解析 json 3 次。

我有一个大约有 40 列的表，对于它来说，这个查询比使用文件格式选项解析 JSON 的更简单的解决方案慢大约 5 倍，但不幸的是不支持on_error=continue option.

copy into HQO_DEVELOPMENT.PUBLIC.DIM_TENANT
from (select $1:id, $1:name, $1:status 
from @HQO_DEVELOPMENT.PUBLIC.DIM_TENANT_STAGE_NN) 
file_format = (type = 'json')

尝试了什么

• 不支持像这样使用嵌套选择: :

copy into HQO_DEVELOPMENT.PUBLIC.DIM_TENANT from 
(select $1:id, $1:name, $1:status from (
select parse_json($1) from
@HQO_DEVELOPMENT.PUBLIC.DIM_TENANT_STAGE))
 on_error = 'continue'

• 在舞台上使用 JSON 类型并在管道上省略：不会解决问题

有没有办法可以受益on_error=continue并且不为每一列解析 JSON？

Snowflake文档说：支持CSV和半结构化文件类型；但是，即使加载半结构化数据（例如 JSON），您也应该将 CSV 设置为文件格式类型（默认值）。您可以使用相应的文件格式（例如 JSON），但转换中的任何错误都会停止 COPY 操作，即使您设置 ON_ERROR 选项来继续或跳过文件。

https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-table.html https://docs.snowflake.net/manuals/sql-reference/sql/copy-into-table.html

另一方面，我发现 on_error 选项适用于 NDJSON 文件，至少当您在阶段级别设置文件类型时。供测试用，

我创建了以下 NDJSON 文件进行测试：

{ "id":1, "name":"Gokhan", "location":"Netherlands" }
{ "id":2, "name":"Hans", location:Germany -- broken json #1 }
{ "id":3, "name":"Joe", "location":"UK" }
{ broken json #2 }
{ "id":4, "name":"Mike", "location":"US" }

我创建了一个文件类型对象，并使用此文件类型创建了一个阶段（您可以更改现有阶段并设置文件类型）：

CREATE FILE FORMAT myformat TYPE = json;
CREATE STAGE mystage FILE_FORMAT = (FORMAT_NAME = myformat);

我将示例 NDJSON 文件上传到此阶段，创建了一个 Snowpipe 来加载它：

CREATE PIPE mypipe AS
COPY INTO mytable 
FROM (SELECT $1:id, $1:name, $1:location FROM @mystage)
ON_ERROR = CONTINUE;

当我刷新管道时，它成功地从文件中加载了“有效”（3）条记录。