Azure Databricks大數(shù)據(jù)構建營 | 小試牛刀，順利搞定流計算

看過美劇《絕命毒師》（Breaking Bad）的童鞋，恐怕都會對「老白」的化學知識和運用這些知識的能力留下深刻印象。畢竟，生活中一些常見的物品，通過對它們的成分進行一定組合往往能制造出很「不一般」的東西。

所以接下來你面對了一個新需求，老板要你實現(xiàn)這樣的功能：一家連鎖藥店，希望通過抽樣的方式檢查止咳糖漿的銷量，如果發(fā)現(xiàn)有人一次性購買10瓶以上就進行實時告警。你準備怎么做？

基于Azure Databricks的流計算就可以輕松實現(xiàn)。一起來看看吧。

本次《Azure Databricks大數(shù)據(jù)構建營》系列文章，將主要圍繞Azure Databricks以及其他配套服務，通過一系列實戰(zhàn)教程告訴大家如何基于Azure云打造完全運行在云端的閉環(huán)大數(shù)據(jù)平臺。

內(nèi)容涵蓋：Azure Databricks的基礎知識，以及如何借助Azure Databricks實現(xiàn)流計算（Structure Streaming）、批處理（Spark SQL）、安全訪問控制以及機器學習能力等。

本系列的第一篇文章概括介紹了有關Azure Databricks的基礎知識，錯過的童鞋可以點擊這里回看。而本篇，將帶領大家小試牛刀，基于Azure Databricks開發(fā)上文提到的那種實時告警系統(tǒng)。準備好，我們這就開始。

流計算架構說明

本篇涉及到的服務全都部署在中國區(qū)Azure上。根據(jù)模擬場景的描述，首先我們可以設計出如下所示的架構：

該解決方案將用到如下的產(chǎn)品和服務：

Azure虛擬機：作為數(shù)據(jù)源，扮演Producer生產(chǎn)者，通過Python代碼模擬客戶購買行為，生成示例數(shù)據(jù)并通過SDK發(fā)送數(shù)據(jù)到Azure EventHub。

Azure EventHub：消息隊列，做上下游生產(chǎn)者消費者服務的解耦，Entity ingestion負責接收Producer發(fā)送的數(shù)據(jù)，Entity alerting接收經(jīng)過Databricks實時計算后的數(shù)據(jù)。

Azure Databricks：訂閱EventHub Entity ingestion作為數(shù)據(jù)源，通過Structure Streaming對數(shù)據(jù)進行實時處理后發(fā)送給Entity alerting。

Azure LogicApp：訂閱EventHub Entity alerting并做郵件實時告警。

完整過程中的數(shù)據(jù)流是這樣的：

1.Producer生產(chǎn)者發(fā)送數(shù)據(jù)

2.EventHub Entity ingestion（Ingestion實體）

3.Databricks Structured Streaming（流計算框架）

4.EventHub Entity alerting（Alerting實體）

5.Logic App

Azure Databrick Structure Streaming的實現(xiàn)

1.Terraform自動化部署

通過Terraform部署的服務組件包括Azure虛擬機、Azure Databricks、EventHub、Logic App，具體的tf文件和變量可參見這里。每項服務Terraform Azure Provider都由Resource支持，具體可參考官方文檔。部署完成后的資源清單如下圖所示，所有資源都部署在中國北二區(qū)域。

2.Producer代碼發(fā)布

模擬的生產(chǎn)者代碼通過虛擬機發(fā)布，通過調(diào)用Azure EventHub的SDK將消息寫入，具體代碼可見這里，幾個重要配置簡單說明下：

azure.eventhub：Azure EventHub SDK包，需要通過pip3 install azure.eventhub來指定安裝。

create_batch()：通過該方法批量發(fā)送數(shù)據(jù)，本次示例以1條消息為1個批次發(fā)送到EventHub Entity ingestion。

CONNECTION_STR：Azure EventHub Endpoint，該連接字符串可以在門戶上Shared access policies的Connection string–primary key中查看。

EVENTHUB_NAME：寫入的EventHub Entity Name。

3.Azure Databricks集群配置及Structure Streaming Notebook的集成

Azure Databricks的創(chuàng)建過程是：首先在Azure上創(chuàng)建一個Databricks實體，然后在此基礎上在實體內(nèi)部創(chuàng)建Workspace以及Cluster，再提交Job。每個Databricks資源都有唯一的ID和Endpoint與之對應，以便能夠進行Restful API調(diào)用，集群通過Databricks門戶創(chuàng)建即可。

本示例將創(chuàng)建1個Driver和2個Worker，共計3個節(jié)點的Standard Cluster，Databricks版本為6.4（包含Apache Spark 2.4.5、Scala 2.11）。如果需要做機器學習相關計算，可啟用集成GPU/ML框架的版本，詳細說明見官方文檔，這里不做贅述。

集群狀態(tài)變?yōu)镽unning就意味著就緒可以使用了。不過在導入Python Notebook之前，需要通過Maven Central安裝com.microsoft.azure:azure-eventhubs-spark庫文件，以便安裝Spark連接Azure EventHub Connector，需要注意庫文件的版本要匹配。

Notebook可以直接在門戶里新建寫入，也可以在VS Code等IDE中編寫完之后發(fā)布。本文采用第二種模式，原因是IDE豐富的插件可以提高效率。具體的Notebook本文不做展示，放在這里，有需要的童鞋可以自行查看。通過import導入后，附上導入后的截圖并做幾點說明：

整個Notebook分為三個階段：

第一階段：從EventHub Entity ingestion讀取Producer寫入的數(shù)據(jù)，通過Streaming DataFrames的spark.readStream()創(chuàng)建。

第二階段：通過DataFrame豐富的函數(shù)做字段篩選，篩選出來我們需要的字段。

第三階段：回寫EventHub Entity alerting，通過Streaming DataFrames的spark.writeStream()流式寫入，注意利用Checkpoint方便任務終止再運行。

當Producer運行起來后，EventHub就會不斷有數(shù)據(jù)寫入，所以能看到Spark的Input Records圖像。對于每一個Job，都能看到對于該任務分配的資源和Spark參數(shù)配置項。

4.Logic APP配置郵件告警

經(jīng)過Azure Databricks的數(shù)據(jù)篩選后，篩選出來的Messages都寫入了EventHub Entity alerting中，此時通過LogicApp來定義一個自動化的工作流來進行郵件告警。具體創(chuàng)建過程選擇Blank，然后自己創(chuàng)建Step即可，當然Azure門戶上的示例模板也可以用來參考，如下圖所示：

第一步訂閱EventHub Entity alerting，第二步集成Outlook郵件接口發(fā)送告警郵件。所以當目標消息被篩選出來之后，LogicApp就按照定義的郵件內(nèi)容（本文是消息內(nèi)容和時間戳）來發(fā)送郵件，發(fā)送郵件的截圖如下：

總結(jié)

總體上，一個通過消息隊列Azure EventHub以及Databricks做流計算處理的示例就完成了。

如果消息生產(chǎn)者Producer不斷產(chǎn)生消息，那么整個任務就會一直運行下去，當出現(xiàn)目標消息的時候就會不斷的持續(xù)告警。在Spark推出Structure Streaming后，也解決了Spark Streaming micro batch的局限性。

本文開頭提及的需求已經(jīng)順利實現(xiàn)！

有關如何借助Azure Databricks實現(xiàn)流計算（Structure Streaming）的內(nèi)容就是這些了。隨后的內(nèi)容中，我們還將介紹如何實現(xiàn)批處理（Spark SQL）、安全訪問控制和機器學習能力。通過這些真實場景中的最佳實踐分享，也能幫助大家更清楚地認識到Azure Databricks的價值。敬請期待！