我建立了一個流媒體分析應用程序使用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隻是跟蹤所有的文件加載,所以確切地知道在哪裏接。