1

情况：每个月都有一个 csv 进入 AWS S3。供应商可以随意添加/删除/修改文件中的列。所以架构是不提前知道的。要求是在 Snowflake 中动态创建一个表并将数据加载到所述表中。Matillion 是我们的 ELT 工具。

这是我到目前为止所做的。

  1. 设置一个 Lambda 来检测文件的到达，将其转换为 JSON，上传到另一个 S3 目录并将文件名添加到 SQS。
  2. Matillion 检测 SQS 消息并将带有 JSON 数据的文件加载到 SF 表的 Variant 列中。
  3. SF Stored proc 采用变量列并根据 JSON 数据中的字段数生成一个表。SF 中的 VARIANT 列仅在其 JSON 数据时以这种方式工作。遗憾的是不支持 CSV。

这适用于 10,000 行。当我使用超过 1GB（超过 10M 行）的完整文件运行它时，就会出现问题。它在运行时因磁盘空间不足错误而使 lambda 作业崩溃。

这些是我到目前为止想到的替代方案：

  1. 将 EFS 卷附加到 lambda，并在上传到 S3 之前使用它来存储 JSON 文件。JSON 数据文件比它们的 CSV 对应文件大得多，我预计 json 文件大约为 10-20GB，因为该文件有超过 10M 行。
  2. Matillion 有一个 Excel 查询组件，它可以在其中获取标题并动态创建表格并加载文件。我在想我可以将 CSV 中的标题行转换为 Lambda 中的 XLX 文件，将其传递给 Matillion，让它创建结构，然后在创建结构后加载 csv 文件。

我在这里还有哪些其他选择？考虑因素包括用于未来大型 CSV 或类似需求的良好可重复设计模式、EFS 成本、我是否充分利用了可用的工具？谢谢！！！

4

4 回答 4

0

为什么要将 CSV 转换为 JSON；CSV 直接加载到表中，无需进行 JSON 特殊需要的任何数据转换，横向展平以将 json 转换为关系数据行；以及为什么不使用 Snowflake Snowpipe 功能将数据直接加载到 Snowflake 中而不使用 Matallion。您可以在加载到 Snowflake 之前将大型 csv 文件拆分为较小的块；这将有助于跨 SF Warehouse 分配数据处理负载。

于 2021-01-27T05:29:36.773 回答
0

这可以使用外部表来实现，其中外部表将映射到单个列，并且分隔符将是换行符。外部表还有一个特殊的虚拟列，可以对其进行处理以动态提取所有列，然后使用存储过程在任何给定时间根据列数创建一个表。有一个有趣的视频讨论了雪花中的这种限制（https://youtu.be/hRNu58E6Kmg）

于 2021-02-08T03:38:51.827 回答
0

我还使用 Matillion 将 SFTP 中的 CSV 文件加载到 Snowflake 中，但不知道架构。

在我的过程中，我在 Snowflake 中创建了一个“临时”表，其中包含 50 个 VARCHAR 列（我们的文件不应超过 50 列）。我们的数据总是包含文本、日期或数字，因此 VARCHAR 不是问题。然后我可以将 .csv 文件加载到临时表中。我相信这也适用于来自 S3 的文件。

这至少会将数据导入 Snowflake。但是，鉴于您的情况，我不确定如何创建“最终”表。我可以想象能够使用标题行，和/或对每列中包含的数据的“类型”进行一些分析，以确定所需的列类型。

但是，如果您可以创建“最终”表，则可以将数据从 temp. 或更改临时表本身。

于 2021-02-08T01:03:58.477 回答
0

为什么不将初始 csv 文件拆分为多个文件，然后以与您当前相同的方式处理每个文件？

于 2021-01-26T22:14:05.270 回答