作者：美的樓宇科技事業(yè)部 先行研究中心智能技術(shù)部

　　美的樓宇科技 IoT 數(shù)據(jù)平臺建設(shè)背景

　　美的樓宇科技事業(yè)部(以下簡稱樓宇科技)是美的集團旗下五大板塊之一，產(chǎn)品覆蓋多聯(lián)機組、大型冷水機組、單元機、機房空調(diào)、扶梯、直梯、貨梯以及樓宇自控軟件和建筑弱電集成解決方案，遠銷海內(nèi)外200多個國家。針對當(dāng)前設(shè)備數(shù)據(jù)量龐大且持續(xù)增長、數(shù)據(jù)呈現(xiàn)半結(jié)構(gòu)化特點的現(xiàn)狀，現(xiàn)有系統(tǒng)僅停留在數(shù)據(jù)存儲和基礎(chǔ)使用層面，缺乏深度挖掘數(shù)據(jù)價值的能力，導(dǎo)致大量潛在信息未被充分利用。因此，迫切需要構(gòu)建一個統(tǒng)一且通用的 IoT 數(shù)據(jù)平臺，平臺不僅要具備高度的彈性和輕量化特性，還應(yīng)具備強大的大規(guī)模數(shù)據(jù)處理能力以及數(shù)據(jù)科學(xué)和 AI 技術(shù)支持，以實現(xiàn)快速的數(shù)據(jù)分析與智能化挖掘，推動樓宇系統(tǒng)的智能化升級，支持節(jié)能、設(shè)備管理和運維等方面的精確決策。我們的 IoT 數(shù)據(jù)平臺建設(shè)基于阿里云 EMR  Serverless Spark ，我們將就IoT數(shù)據(jù)平臺建設(shè)技術(shù)選型上的一些思考，以及 Spark 技術(shù)棧尤其是場景應(yīng)用實踐做一下分享。

　　Lakehouse 架構(gòu)

　　樓宇科技通過阿里云EMR Serverless Spark，實現(xiàn)了數(shù)據(jù)與 AI技術(shù)的有效融合，并結(jié)合EMR Serverless StarRocks搭建了Lakehouse 平臺。該平臺核心部分如下：

　　首先，上游設(shè)備或傳感器數(shù)據(jù)通過Serverless Spark提交Streaming作業(yè)，實時以Apache Hudi格式寫入數(shù)據(jù)湖，湖表元數(shù)據(jù)同步至DLF，以保持數(shù)據(jù)的實時性。

　　接著，采用天級調(diào)度執(zhí)行Hudi分區(qū)數(shù)據(jù)的Compaction，并使用 Z-order 來優(yōu)化數(shù)據(jù)布局，實現(xiàn)了10倍以上的查詢加速。同時，DLF的鎖機制確保了實時寫入與異步湖表任務(wù)的并發(fā)事務(wù)管理，為作業(yè)穩(wěn)定性、數(shù)據(jù)一致性提供了保障。

　　此外，還通過 Serverless Spark構(gòu)建了數(shù)據(jù)Medallion架構(gòu)，從加載的源始數(shù)據(jù)開始(Bronze)，經(jīng)過清洗轉(zhuǎn)化為明細數(shù)據(jù)(Silver)，然后根據(jù)不同業(yè)務(wù)需求將明細層數(shù)據(jù)轉(zhuǎn)化為高質(zhì)量的指標數(shù)據(jù)(Gold)，為上層業(yè)務(wù)系統(tǒng)提供支持。

　　在AI應(yīng)用方面，樓宇科技通過Serverless Spark  PySpark 任務(wù)，并基于PyArrow UDF調(diào)用自研算法實現(xiàn)了千億級別數(shù)據(jù)在百萬級維度的聚合，推動了Data + AI技術(shù)在實際業(yè)務(wù)中的應(yīng)用。最后，處理后的指標數(shù)據(jù)從數(shù)據(jù)湖中被加載到StarRocks中，為上層應(yīng)用提供Dashboard和報表支持，提升了數(shù)據(jù)的可視化和決策能力。

　　以下架構(gòu)圖展示了如何利用Serverless Spark結(jié)合開源湖格式Hudi、ML/AI的多種工具庫，以及阿里云 DLF 統(tǒng)一湖倉管理平臺，實現(xiàn)高效的數(shù)據(jù)處理和AI賦能，使用Serverless StarRocks實現(xiàn)極速數(shù)據(jù)分析，為業(yè)務(wù)應(yīng)用帶來顯著的提升。

　　選擇  Spark 技術(shù)棧

　　在數(shù)據(jù)平臺計算引擎層技術(shù)選型上，前期的架構(gòu)選型我們做了很多的調(diào)研，綜合各個方面考慮，希望選擇一個成熟且統(tǒng)一的平臺：既能夠支持數(shù)據(jù)處理、數(shù)據(jù)分析場景，也能夠很好地支撐數(shù)據(jù)科學(xué)場景。加上團隊成員對 Python 及 Spark 的經(jīng)驗豐富，所以，從一開始就將目標鎖定到了 Spark 技術(shù)棧。

　　為什么選擇阿里云EMR Serverless Spark

　　EMR Serverless Spark 解決了我們什么痛點

　　1. 自建集群 POC 測試需要花費大量的成本，周期也比較長;

　　2.針對千億級別的IOT設(shè)備上報數(shù)據(jù)，引擎性能非常關(guān)鍵。對原始數(shù)據(jù)做一輪點位提取(t+1處理)，用于后續(xù)數(shù)據(jù)開發(fā)和分析，每日的點位提取需要在短時間內(nèi)運行大量資源對湖原始數(shù)據(jù)進行查詢和處理;

　　3. 需要完善的Spark 生態(tài)，來實現(xiàn)全鏈路數(shù)據(jù)流轉(zhuǎn)，來滿足批、流、交互式、機器學(xué)習(xí)等不同場景需求;

　　4. 彈性計算能力，需要一次性支持大規(guī)模計算，縮短數(shù)據(jù)使用延遲。多聯(lián)機能耗運行月度報告生成的過程中，每月5號之前需要大量資源去生成上月的月度報告指標;

　　5. Data+AI場景的支持能力。

　　成本相比過去架構(gòu)提升

　　1. 不同場景下的整體性能提升50%以上

　　2. 綜合成本下降30%左右

　　IoT 數(shù)據(jù)鏈條

　　我們接入的 IoT 數(shù)據(jù)分為兩部分，歷史存量數(shù)據(jù)和實時數(shù)據(jù)。目前，歷史存量數(shù)據(jù)是通過 Spark SQL 以天為單位從不同客戶關(guān)系數(shù)據(jù)庫批量導(dǎo)入 Hudi Lake 表中;實時數(shù)據(jù)通過 IoT 平臺采集到云 Kafka ，經(jīng)由 Spark Structured Streaming 消費后實時寫入到 Hudi Lake 表中。在這個過程中，我們將實時數(shù)據(jù)和歷史數(shù)據(jù)都 sink 到同一張 Hudi 表里，這種批流一體操作可大大簡化我們的 ETL 流程(參考后面的案例部分)。數(shù)據(jù)管道下游，我們對接數(shù)據(jù)分析及數(shù)據(jù)科學(xué)工作流。

　　IoT 數(shù)據(jù)采集：從 Little Data 到 Big Data

　　作為 IoT 場景的典型應(yīng)用，美的暖通最核心的數(shù)據(jù)均來自 IoT 終端設(shè)備。在整個 IoT 環(huán)境下，分布著無數(shù)個終端傳感器。從小的維度看，傳感器產(chǎn)生的數(shù)據(jù)本身屬于 Small Data(或者稱為 Little Data)。當(dāng)把所有傳感器連接成一個大的 IoT 網(wǎng)絡(luò)，產(chǎn)生自不同傳感器的數(shù)據(jù)經(jīng)由 Gateway 與云端相連接，并最終在云端形成 Big Data 。

　　在我們的場景下，IoT 平臺本身會對不同協(xié)議的數(shù)據(jù)進行初步解析，通過定制的硬件網(wǎng)絡(luò)設(shè)備將解析后的半結(jié)構(gòu)化 JSON 數(shù)據(jù)經(jīng)由網(wǎng)絡(luò)發(fā)送到云 Kafka。云 Kafka 扮演了整個數(shù)據(jù)管道的入口。

　　數(shù)據(jù)入湖：Hudi

　　IoT 場景下的數(shù)據(jù)有如下幾個特點：

　　時序數(shù)據(jù)：傳感器產(chǎn)生的數(shù)據(jù)記錄中包含時間相關(guān)的信息，數(shù)據(jù)本身具有時間屬性，因此不同的數(shù)據(jù)之間可能存在一定的相關(guān)性。利用 as-of-join 將不同時間序列數(shù)據(jù) join 到一起是下游數(shù)據(jù)預(yù)測分析的基礎(chǔ)

　　數(shù)據(jù)的實時性：傳感器實時生成數(shù)據(jù)并以最低延遲的方式傳輸?shù)綌?shù)據(jù)管道,觸發(fā)規(guī)則引擎，生成告警和事件，通知相關(guān)工作人員。

　　數(shù)據(jù)體量巨大：IoT 網(wǎng)絡(luò)環(huán)境下遍布各地的成千上萬臺設(shè)備及其傳感器再通過接入服務(wù)將海量的數(shù)據(jù)歸集到平臺

　　數(shù)據(jù)協(xié)議多樣：通常在 IoT 平臺接入的不同種類設(shè)備中，上傳數(shù)據(jù)協(xié)議種類多樣，數(shù)據(jù)編碼格式不統(tǒng)一

　　數(shù)據(jù)半結(jié)構(gòu)化:  不同設(shè)備包含不同的屬性，基于JSON 結(jié)構(gòu)把所有IoT模型抽象為JSON 字符串

　　IoT 數(shù)據(jù)上述特點給數(shù)據(jù)處理、數(shù)據(jù)分析及數(shù)據(jù)科學(xué)等帶來了諸多挑戰(zhàn)，慶幸的是，這些挑戰(zhàn)借助 Spark 和 Delta Lake 都可以很好地應(yīng)對。Hudi Lake 提供了 ACID 事務(wù)保證，支持增量更新數(shù)據(jù)表以及流批同時寫數(shù)據(jù)。借助 Spark Structed Streaming 可以實現(xiàn) IoT 時序數(shù)據(jù)實時入湖。

　　以下是 Hudi Lake 經(jīng)典的三級數(shù)據(jù)表架構(gòu)。具體到樓宇科技 IoT 數(shù)據(jù)場景，我們針對每一層級的數(shù)據(jù)表分別做了如下定義：

　　Bronze 表：存儲原生數(shù)據(jù)(Raw Data)，數(shù)據(jù)經(jīng)由 Spark Structed Streaming 從 Kafka 消費下來后 Append/Upsert 進 Hudi Lake 表，該表作為唯一的真實數(shù)據(jù)表  (Single Source of Truth)

　　Silver表：該表是在對 Bronze 表的數(shù)據(jù)進行加工處理的基礎(chǔ)上生成的中間表，在美的暖通的場景下，數(shù)據(jù)加工處理的步驟涉及到一些復(fù)雜的時序數(shù)據(jù)計算邏輯，這些邏輯都包裝在了 Pandas UDF 里提供給 Spark 計算使用

　　Gold 表：Silver 表的數(shù)據(jù)施加 Schema 約束并做進一步清洗后的數(shù)據(jù)匯入 Gold 表，該表提供給下游的 Ad Hoc 查詢分析及數(shù)據(jù)科學(xué)使用

　　數(shù)據(jù)分析：Ad-Hoc 查詢 & 實時分析

　　我們內(nèi)部在開源 Superset 基礎(chǔ)上定制了內(nèi)部版本的 SQL 查詢與數(shù)據(jù)可視化平臺，通過StarRocks Lake Catalog實現(xiàn)對湖數(shù)據(jù)查詢。借助 Superset ，數(shù)據(jù)分析師及數(shù)據(jù)科學(xué)家可以快速高效的對 Hudi Lake 表進行數(shù)據(jù)探索。

　　StarRocks主要應(yīng)用于BI報表分析平臺 、實時大屏(如設(shè)備實時跟蹤場景)，通過Serverless StarRocks可大大提高對數(shù)據(jù)湖的分析和查詢性能，相較于Trino等查詢性能有3-5倍性能提升。且利用物化視圖可以對實時寫入數(shù)據(jù)進行再次近實時加工和處理，滿足大屏分析等實時數(shù)據(jù)展示、進一步提升查詢性能、降低資源使用。

　　數(shù)據(jù)科學(xué)：Jupyter 交互式開發(fā)

　　樓宇能耗優(yōu)化與設(shè)備故障診斷預(yù)測是樓宇科技IoT 大數(shù)據(jù)平臺建設(shè)的兩個主要業(yè)務(wù)目標。在 IoT 數(shù)據(jù)管道下游，需要對接機器學(xué)習(xí)平臺�，F(xiàn)階段為了更快速方便地支撐起數(shù)據(jù)科學(xué)場景，Serverless Spark 支持對接在數(shù)據(jù)科學(xué)場景下更友好的 Jupyter Notebook ，通過在 Jupyter 上使用 PySpark ，可以將作業(yè)運行到Serverless Spark上;對于有周期性執(zhí)行的作業(yè)，也可以借助 Apache Airflow 對作業(yè)進行調(diào)度。同時，考慮到機器學(xué)習(xí)模型構(gòu)建、迭代訓(xùn)練、指標檢測、部署等基本環(huán)節(jié)，我們也在探索 MLOps ，目前已概念驗證通過OSS+MLflow+Serverless Spark

　　Hudi Lake 數(shù)據(jù)入湖(批流一體)

　　query = (

　　df.writeStream

　　.outputMode("append")

　　.options(**hudi_options)

　　.format("hudi")

　　.option("path", table_oss_path)

　　.option("checkpointLocation", streaming_checkpoint_location)

　　.trigger(availableNow=True)

　　.start()

　　)

　　湖表管理

　　Compaction & Z-Ordering

　　通過Spark Streaming實時的將數(shù)據(jù)寫入到Hudi湖存儲上能夠提升數(shù)據(jù)的新鮮度，但同時也產(chǎn)生大量的小文件影響下游系統(tǒng)的查詢性能。另外，對于查詢模式相對固定的Hudi表，我們也通過Z-Order來優(yōu)化數(shù)據(jù)布局，再借助Data-Skipping能力能夠進一步提高查詢性能。同時由于Z-Order使得局部數(shù)據(jù)結(jié)構(gòu)相似，也使得以Parquet格式存儲時有更大的壓縮效果，降低了存儲成本。

　　美的樓宇客戶IoT數(shù)據(jù)以天為維度進行分區(qū)管理，數(shù)據(jù)實時注入到特定的天級分區(qū)內(nèi)，因此我們通過EMR Serverless Spark產(chǎn)品以T+1的方式對T分區(qū)內(nèi)的數(shù)據(jù)進行帶有Z-Order的Compaction實現(xiàn)了高效的Hudi表的文件管理，有效的提升了查詢性能。

　　call run_clustering(

　　table => '{db_name}.{table_name}',

　　op => 'scheduleAndExecute',

　　order => 'device_id',

　　order_strategy => 'z-order',

　　predicate => '({predicate})',

　　show_involved_partition => false,

　　options => "{options}"

　　);

　　Clean

　　Hudi Lake支持事務(wù)提交提供了多版本、TimeTravel等豐富的功能，但也使得歷史的過期的文件依然保留在文件系統(tǒng)中造成存儲的浪費。我們也基于EMR Serverless Spark實現(xiàn)了天級調(diào)度Clean作業(yè)來定期清除不需要的數(shù)據(jù)文件，避免存儲資源浪費。

　　總結(jié)與展望

　　我們基于阿里云 EMR Serverless Spark技術(shù)�？焖贅�(gòu)建了 IoT 數(shù)據(jù)處理平臺，Serverless Spark全托管免運維、自研 Fusion 引擎，內(nèi)置高性能向量化計算和 RSS 能力，相比開源版本3倍以上的性能優(yōu)勢以及計算/存儲分離的架構(gòu)，為我們節(jié)省了總體成本。同時，EMR Serverless Spark自身提供的豐富特性，也極大提升了我們數(shù)據(jù)團隊的生產(chǎn)力，為數(shù)據(jù)分析業(yè)務(wù)的快速開展交付奠定了基礎(chǔ)。未來，美的樓宇科技希望與阿里云 EMR 團隊針對 IoT 場景輸出更多行業(yè)先進解決方案。