自动将文件从 s3 加载到雪花

Question

在 s3 存储桶中，每天都有新的 JSON 文件被转储，我必须创建解决方案，当它到达时选择最新的文件 PARSE JSON 并将其加载到 Snowflake Datawarehouse。有人可以分享你的想法我们怎样才能实现

Answer 1

根据您的需要，有多种方法可以做到这一点。我建议创建一个事件来触发 lambda 函数。

https://docs.aws.amazon.com/lambda/latest/dg/with-s3.html

另一种选择可能是在文件到达 s3 时创建 SQS 消息，并让 ec2 实例轮询队列并根据需要进行处理。

https://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html https://boto3.amazonaws.com/v1/documentation/api/latest/guide/sqs-example-long-polling.html

编辑：这里更详细地解释了如何从 s3 创建事件和触发 lambda 函数。文档由 Snowflake

提供

https://docs.snowflake.net/manuals/user-guide/data-load-snowpipe-rest-lambda.html

Answer 2

查看 Snowpipe，它可以让您在系统内执行此操作，使其（可能）更容易。

Answer 3

有一些方面需要考虑，例如它是批处理数据还是流数据，您是否要重试加载文件以防数据或格式错误，或者您是否想让它成为一个通用过程以便能够处理不同的文件格式/文件类型（csv/json）和阶段。 在我们的例子中，我们使用 Python 和 Luigi 构建了一个通用的 s3 到 Snowflake 负载，并且还使用 SSIS 实现了相同的功能，但仅适用于 csv/txt 文件。

Answer 4

在我的例子中，我有一个 python 脚本，它使用 boto 获取有关存储桶的信息。

检测到更改后，我调用 SnowPipe 上的 REST 端点插入文件。

定相：

• 检测 S3 变化
• 获取 S3 对象路径
• 在 S3 中解析内容并转换为 CSV（相同的存储桶或其他 snowpipe 可以连接）
• 调用 SnowPipe REST API

你需要什么：

• 使用 public 键创建用户
• 使用 AWS 凭证在 SnowFlake 上创建您的舞台以访问 S3
• 使用您的用户角色在 Snowflake 上创建管道
• 签署 JWT

我还尝试了 TOS BigData 的 Talend 作业。

希望对您有所帮助。