问题标签 [snowflake-pipe]

我们已经使用 Snowpipe 大约 10 个月了，我们最近遇到了一个案例，其中部分文件已stage上传到相应的雪花表，并且未检测到任何未来的文件。验证底层证券stage处于pipe有效状态。

假设暂存位置是 s3://<some_bucket>/some/path 并且有 5 个文件 file1.csv、file2.csv、file3.csv、file4.csv、file5.csv

输出表明检测到所有 5 个文件，并且计数与预期一致。但 file4.csv 和 file5.csv 从未被摄取。

不显示复制历史，这使我们怀疑表/管道是否处于不确定状态以及是否有解决方法。

注意：该copy_history命令适用于数据库中的其他表。

我正在尝试使用 Snowflake 站点中指出的 Snowpipe rest api：
https ://docs.snowflake.com/en/user-guide/data-load-snowpipe-rest-apis.html#data-file-ingestion

我在这里找到了一个 python 示例，我的代码和步骤几乎相同：
https ://community.snowflake.com/s/article/Connect-to-Snowpipe-Rest-API-by-using-JWT-Token

我检查了https://jwt.io/#debugger中的令牌，它是一个有效的 jwt 令牌。

然而，Snowpipe api 总是响应：

我错过了什么吗？

我使用以下步骤创建了密钥：
https ://docs.snowflake.com/en/user-guide/data-load-snowpipe-rest-gs.html#step-3-configure-security-per-user

我也试过thise其他python代码（和其他的），但有同样的错误：
https ://docs.snowflake.com/en/user-guide/data-load-snowpipe-rest-load.html#sample-program -for-the-python-sdk

我有一个似乎设置得很好的管道，但它就是不工作。

我跑了

我得到了越来越多的人numOutstandingMessagesOnChannel

有人可以解释一下这是什么意思吗？

是要处理的行吗？我预计这个数字会下降吗？有没有太高的数字？

有什么/某种方法可以跟踪它为什么/什么时候上涨吗？

该文件仅说，

我想将所有管道放入与模式匹配的雪花模式中。

您可以显示与此处所示模式匹配的管道。

例子：show pipes like '%NAME_LIKE_THIS%' in MY_DB.MY_SCHEMA

但是， drop pipe似乎不存在类似的功能。

我正在考虑创建一个存储过程，它将接收pattern和schema参数，并遍历每个并删除，但我希望有更好/更简单的方法。

先感谢您。

我创建了一个包含多个 DML 的文件。然后我从SNOWSQL客户端调用该文件。

是否可以通过snowpipe这样做，这样我就不需要每次都手动调用文件？

下面是我在snowsql客户端使用的脚本：

我正在阅读本用户指南，其中提到“通常”雪管需要 1 分钟来加载数据。在我的实验中，我发现它总是需要一分钟。这 1 分钟的延迟是从哪里来的？感觉每分钟都在进行一些批处理。是否有某个地方可以进一步降低它。

我使用了如下复制命令：

从@%test2 file_format = (format_name = 'CSV') on_error = 'CONTINUE' 复制到 test2；

我的文件在数字字段中包含一些字符数据（对于所有记录），因此复制结果为 LOAD_FAILED，我可以使用以下查询获取失败记录（在这种情况下，所有记录都是失败记录）：

select * from table(validate("TEST2", job_id=>'对应的JOB ID'));

我还尝试提供无效日期，但仍然从上述查询中获得了所有不良记录。

现在我尝试了如下复制命令：从 @%test2 中选择 $1,to_date($2,'YYYYDDD') 复制到 test2(test1,test2) file_format = (format_name = 'CSV') on_error = 'CONTINUE';

复制结果又是 LOAD_FAILED 但我现在没有从下面的查询中得到任何失败记录： select * from table(validate("TEST2", job_id=>'Corresponding JOB ID'));

这是否仅适用于常规副本，而 Copy 中没有任何转换功能或有任何其他原因？

在看到下面 Mike 的回复后添加一个示例：文件数据：1,2018-1-34 2,2/3/2016 3,2020124

表-> 创建表 test2（test1 编号，test2 日期）

复制到 test2(test1,test2) from (select $1,to_date($2,'YYYYDD') from @%test2) file_format = (format_name = 'CSV') on_error = 'CONTINUE';

第一条和第三条记录在验证查询中可用。在这种情况下，只有第二条记录不存在。有点奇怪。（所有三个记录都在复制中失败）。

正如 Mike 在下面的评论中所说，验证不适用于复制中的转换数据，但为什么在这种情况下它提供两条记录。它应该枯萎不提供任何东西或全部？

我是雪花的新手。我想知道如何为雪花的数据构建数据目录？

我找不到这方面的文件。我可以找到这个帖子。但它似乎没有使用目录功能。

我想从雪花中获取转换后的数据到我的 postgres 数据库。为了实现这一点，我想将数据从雪花移动到卡夫卡，然后从卡夫卡到 postgres。无法弄清楚如何将数据从雪花移动到卡夫卡。

命令状态的Snowkflake 文档COPY INTO（用于 COPY 选项）

ON_ERROR = 继续 | 跳过文件 | SKIP_FILE_num | SKIP_FILE_num% | ABORT_STATEMENT

继续加载文件。COPY 语句针对每个数据文件遇到的最多一个错误返回错误消息。请注意，ROWS_PARSED 和 ROWS_LOADED 列值之间的差异表示包含检测到的错误的行数。但是，这些行中的每一行都可能包含多个错误。要查看数据文件中的所有错误，请使用 VALIDATION_MODE 参数或查询 VALIDATE 函数。

但对我来说，它似乎不服从，因为我看到了默认值，即 SKIP_FILE 被应用，因为文件中的任何错误都会被跳过。

选项是否ON_ERROR=CONTINUE适用于 PIPE？

注意：该文件是一个 NDJSON 文件。