使用Python和Snowflake处理不断变化的源模式问题

问题陈述：

一个拥有大量快速消费品的组织，在印度各地的多个城市开展活动。这次活动的主要目的是推广产品，并从现有客户和潜在客户那里收集反馈。

数据以excel文件的形式收集并加载到云端的一个集中位置。尽管excel样本文件已与活动人士共享，但如果他们认为收集到的数据可能相关，他们会增加新的栏目。在这样的场景中，所有输入源文件的结构可能相同，也可能不同。
因此，大多数情况下，每个新的源文件都有一个新的模式，即添加一些列，或删除一些列。在这种情况下，读取输入数据并将其发送到管道中的下一个步骤是不可能的，ETL流程在第一步就失败了。

但是，这些数据非常重要，需要进一步处理，因为组织希望获得有用的业务见解。

的方法:

由于我们的原始数据每次都在波动，这将导致表结构的变化/为了避免这种情况，我们使用简单的编程语言，如Python来读取源文件并将其转换为JSON文件。

现在可以轻松地将文件中的JSON数据转储到Snowflake数据库中的阶段表中。

要将源文件转换为JSON文件，我们必须通过集中处理数据仓库报告所需的特定列，将其转储到雪花的staging表中。

JSON文件可以通过两个步骤加载到雪花:

首先，使用PUT命令，其中我们需要将数据文件上传到雪花内部阶段。
其次，使用COPY INTO将文件从内部阶段加载到雪花表。

为了完成这两步过程，我们需要完成一些先决条件:

为JSON数据创建目标关系表-

创建或替换表JSON_data(
/*需要的列*/
）;
创建命名文件格式，将文件分隔符设置为none，并将记录分隔符设置为新行字符-

当加载JSON文件或任何半结构化文件时，我们应该设置CSV作为文件格式类型，如果我们使用JSON作为文件格式，转换中的任何错误都将停止COPY操作

创建引用文件格式对象的临时内部阶段。

现在使用PUT命令，我们将把数据转储到内部存储(对于windows):
Put file://%TEMP%/(JSON文件名)@(临时内部阶段)

在COPY语句中创建一个查询，标识从其中加载的数据文件中的一组编号的列。

因此，在完成这两步过程后，我们可以在转储源数据后在snowflake中获得最终的staging表。
因此，在使用一个简单的select语句将不断变化的源数据放入雪花之后，我们可以获得所需的列，这将有助于减少映射列的工作，从而减少处理时间和ETL工具的负担。

结论:

由于我们谈论的是与化妆品和日常护理产品等快消品活动相关的不断变化的数据，因此被处理和加载到数据库的数据量是巨大的。

然而，由于我们使用雪花，处理时间比任何on-prem数据库都要短得多，使用雪花有很多优点，比如:
1.存储容量——这是一个基于云的存储blob，即云对象存储解决方案，有助于存储大量的非结构化数据。雪花可以在微软Azure上运行
2.多云——我们也可以在其他流行的平台上托管雪花。
3.性能调优——它是用户友好的，并帮助用户组织数据和一个具有最佳性能的高响应平台。
4.星型模式——雪花是一种星型模式设计，它只是使它更快，更好地优化。

在这样的场景中使用雪花可以帮助我们截断ETL工具中映射每一列的工作，同时构建通过管道转换数据的过程。