我建立了一个流媒体分析应用程序使用SQL和δ住表吗

通过理查德·汤姆林森

2022年5月19日, 在产品

分享这篇文章

计划我的旅行

我想带你到我如何使用砖的旅程最近推出了三角洲生活表产品构建端到端分析应用程序使用实时数据SQL-only技能。

我加入了砖作为一个产品经理2021年11月初。显然我在公司还是一个新手,但是我已经在数据仓库工作,BI,自90年代中期业务分析。我已经建立了相当一部分数据仓库和数据集市在我的时间(金伯尔或Inmon,随你挑吧),使用了几乎每一个ETL和BI工具下太阳在同一时间或另一个。

我不是一个工程师按照今天的标准数据。我很了解SQL,但我更多的是遥控器编码器。我的技术经验和工具Informatica现在,Trifacta(的一部分Alteryx),DataStage等与语言,像Python和Scala。我的角色我认为,更像我们的朋友印度生物技术部实验室将调用一个分析工程师与一个数据工程师。

所以这一切为背景,为了学习尽可能多的砖产品我可以在公司(鉴于我新手状态),我踏上旅程来构建应用程序,我不希望它只是另一个无聊的静态BI仪表板。我想构建一些更类似于一个生产应用与实际实时数据。

因为我住在芝加哥,我要用分配的自行车数据。我看过很多演示使用他们静态数据集但几乎没有使用他们实时api。这些api跟踪“活”站状态(例如#自行车可用,#码头,等等)的所有842个站。鉴于自行车租赁是如此依赖于天气,我会加入这个数据与实时天气信息在每个车站使用OpenWeather api。这样我们可以看到残酷的芝加哥的冬天对分配的影响自行车的使用。

源数据捕捉和消化

我们的数据源分配的自行车和OpenWeatherapi,我需要做的第一件事是弄清楚如何捕捉这些数据是可用的湖在我们云数据(例如ADLS对于我来说,我的砖工作空间中运行Azure)。

有很多数据摄取工具我可以选择这一任务。很多,比如Fivetran是通过我们在几个点击合作伙伴联系生态系统。但是,为了简单起见,我刚刚创建的3简单的Python脚本调用api,然后把结果写进湖的数据。

一旦建立和测试,我作为两个不同的配置脚本砖的工作。

第一份工作得到了每分钟实时站状态,返回一个JSON文件的当前状态1200左右分配自行车站在芝加哥。一个示例有效载荷的样子这。管理数据量和数量的文件将会关注的。我们检索每个站的状态,每一分钟。这将收集1440 JSON文件(60文件/小时* 24小时)~ 1.7米每天新行。按照这个速度,一年的数据给我们630 ~ 520 k JSON文件和~行处理。

第二个工作包括两个任务,每小时运行一次。

第一个任务检索的描述性信息等每一站的名称,类型,lat,长,等等。这是一个典型的缓慢变化维度的数据仓库而言,因为我们不希望这些信息变化频繁。即便如此,我们将更新这个数据每小时,以防它;例如,一个新的站可能会在线,或现有的可以被更新或停用。查看样本有效载荷在这里。

第二个任务的工作然后获取实时天气信息为每个1200左右。这是一个有效载荷的电台。我们所说的API使用它的纬度/经度坐标。因为我们将调用OpenWeather API为每个站我们将每天得到28800个文件(1200 * 24)。一年给我们推断~ 10.5 m JSON文件管理。

这些脚本已经运行了一段时间了。2022年1月4日,我开始和他们愉快地创建新文件在我数据湖。

实现我的“简单”演示实际上是相当复杂的

知道我现在需要融合和转换这些数据,潜在的卷上做数学,看数据样本,这就是我开始出汗。我咬掉超过我能咀嚼吗?有几件事情让这个具有挑战性的和普通的“静态”指示板:

我不知道如何管理数以千计的新每天不断到达的JSON文件。我也想捕捉几个月的数据历史趋势。这是数以百万计的JSON文件管理!
我如何构建一个实时ETL管道的快速分析数据准备好了吗?我的源数据是原始JSON和需要清洗,改变,加入与其他来源,和聚合分析性能。将会有大量的步骤和依赖关系在我的管道需要考虑。
我怎么处理增量加载?我显然不能从头开始重建我的表数据时不断涌入数据湖和我们想要构建一个实时的仪表板。所以我需要找到一个可靠的方法来处理不断移动的数据。
OpenWeather JSON模式是不可预测的。我很快了解到,可以随时间变化的模式。例如,如果它不下雪,你没有得到雪度量返回的有效载荷。如何设计一个目标模式时不能预测源模式! ?
如果我的数据管道失败,会发生什么?我怎么知道当它失败,我怎么重新启动它的地方重新开始?我怎么知道哪一个JSON文件已经处理,哪些没有?
在我的仪表板查询性能怎么样?如果他们是实时指示板需要时髦。我不能有未完成的查询时,新的数据不断流入。化合物,我会很快处理数以亿计的行(如果不是数十亿)。我怎么性能调优的?如何优化和维护我的源文件?的帮助!

好的,我现在就停止。我有点慌张就写这个列表,我相信有一百个其他小障碍跳过。我甚至有时间去构建这个吗?也许我应该看一些别人的视频,叫它一天做什么?

不。我将继续!

减压与达美住表

好,那么接下来,如何编写一个实时ETL管道。好吧,不“真实”的实时。我称之为接近实时的——我相信90%的人真正的意思是当他们说他们需要实时。给我只把数据api每一分钟,我不会比这更新鲜的数据分析应用程序。这是对一个监控这样的用例。

砖最近宣布的完整的可用性三角洲生活表(又名DLT)。DLT恰好是适合这个,因为它提供了“一个简单的声明式的方法来构建可靠的数据管道而自动管理大规模基础设施,所以数据分析师和工程师可以花更少的时间在工具和专注于从数据获得价值。”对我来说,这听起来不错!

DLT还允许我构建管道SQL这意味着我可以保持SQL-only目标。的价值,它还允许您构建管道Python如果你选择——但这不是为我。

重大胜利是DLT允许您编写声明ETL管道、意义而不是手动低级ETL逻辑,我可以花时间在“是什么”,而不是“如何”去做。与DLT,我只是指定如何变换和应用业务逻辑,而DLT自动管理管道内的所有依赖项。这样可以确保我的管道中的所有表正确填充,以正确的顺序。

这是我想构建一个伟大大奖章架构简化变化数据捕获相同的数据和启用多个用例,包括那些涉及数据科学和机器学习,去的众多原因之一Lakehouse在数据仓库。

其他大型DLT的好处包括:

数据质量检查来验证记录流经管道时基于预期(规则)我组
自动错误处理和恢复,所以如果我的管道向下,它可以恢复!
开箱即用监控所以我可以查看实时管道卫生统计和趋势
单击部署到生产和回滚选项,允许我我应该遵循CI / CD模式选择

还有,DLT在批处理或连续工作!这意味着我可以保持我的管道“总是”,不需要知道复杂的流处理或如何实现恢复逻辑。

好的,所以我觉得这个地址从一节我的大部分问题。我能感觉到我的压力已经消退。

快速浏览DLT的SQL代码

这都是什么样子的呢?你可以下载我的DLT SQL笔记本在这里如果你想要动手;死很简单,但我将指导您完成赛事集锦。

首先,我们构建出青铜表图案架构。表中的这些表仅仅代表了原始JSON格式。在这个过程中,我们将JSON数据三角洲湖格式,这是一个开放的格式存储层,提供可靠性、安全性和性能数据湖上。我们不是真的改变数据在这一步。这里有一个例子为一个表:

首先,请注意,我们已经定义了这是一个“流”表。这意味着表将自动支持更新基于连续到达的数据,而不必再计算整个表。

您还会注意到,我们也使用自动加载程序(cloud_files)阅读原始JSON对象存储(ADLS)。自动加载器是一个关键的一部分,这个管道,并提供一个无缝的方式以低成本和延迟加载原始数据以最小的DevOps的努力。

自动加载程序逐步过程新文件,因为他们降落在云存储所以我不需要管理任何状态信息。它有效地跟踪新文件到达时利用云服务,而无需一个目录中的所有文件列表。这是可伸缩的即使有成千上万的文件目录。它也非常容易使用,并将自动设置所需的所有内部通知和消息队列服务增量处理。

它还处理模式推理和演化。你可以阅读更多在这里但简而言之,这意味着我不需要提前知道JSON模式,并将优雅地处理“进化”模式随着时间的推移,没有失败我的管道。适合我的OpenWeather API载荷——另一个压力因素消除。

一旦我有我所有的青铜级别定义表我可以开始做一些真正的ETL工作清理原始数据。这里有一个例子我如何创建一个“银”奖章表:

你会发现一些很酷的东西。首先,它是另一个流表,所以一旦抵达源表的数据(raw_station_status),它会流到这个表。

接下来,请注意,我有设定一个规则station_id不是零。这是一个例子DLT期望或数据质量约束。我可以声明这些我喜欢的。一个期望由一个描述,一个规则(不变的),和一个动作记录失败时规则。上面我决定放弃行从表中如果遇到一个空station_id。三角洲捕捉住表管道事件日志,所以我可以轻松地监视诸如频率规则触发帮我评估的质量数据,并采取适当的行动。

我还添加了一个注释和表属性,因为这是一个最佳实践。谁不喜欢元数据?

最后,您可以释放SQL转换数据的全部威力如何。注意到我爆炸我的JSON为多个行和执行一大堆datetime转换为下游报告目的。

处理缓慢变化维度

上面的示例概述了ETL加载一个事务性或逻辑事实表。所以下一个常见的设计模式,我们需要处理的概念缓慢变化维度(SCD)。幸运的是DLT处理这些!

砖刚刚宣布的DLT支持常见的疾控中心模式用一个新的声明应用为SQL和Python特性变化。这个新功能允许ETL管道容易检测源数据更改,并将它们应用到整个lakehouse数据集。DLT过程数据变化逐步进入三角洲湖,萎靡不振的记录插入、更新或删除在处理疾病预防控制中心事件。

我们station_information何时使用的数据集是一个很好的例子。

而不是简单地添加,我们更新行如果它已经存在(基于station_id)或如果它不插入一个新行。我甚至可以删除记录使用应用时删除条件但我学会了很久以前,我们从来没有在数据仓库中删除记录。这是作为一个分类化合物1型。

部署数据管道

我只有铜和银表创建管道到目前为止,但没关系。我可以创建黄金级别表pre-aggregate提前我的一些数据使我的报告能够跑得更快,但我不知道如果我需要他们,可以随时添加。

目前部署的数据管道是这样的:

3枚铜牌(生)表、一个中间视图(需要一些JSON体操),和3银表,准备报告。

部署管道很容易。我刚刚把我所有的SQL扔进一个笔记本和创建了一个连续的(与触发)DLT管道。由于这是一个演示应用程序我还没有把它搬到生产,但有一个按钮,我可以开发和生产模式之间切换来改变管道底层基础设施上运行。在开发模式中,我可以避免重启自动重试和集群,但是开关这些生产。我也可以启动和停止这个管道一样我想要的。DLT只是跟踪所有的文件加载,所以确切地知道在哪里接。