背景

隨著數據時代的不斷發展，數據量爆發式增長，數據形式也變的更加多樣。傳統數據倉庫模式的成本高、響應慢、格式少等問題日益凸顯。于是擁有成本更低、數據形式更豐富、分析計算更靈活的數據湖應運而生。

數據湖作為一個集中化的數據存儲倉庫，支持的數據類型具有多樣性，包括結構化、半結構化以及非結構化的數據，數據來源上包含數據庫數據、binglog 增量數據、日志數據以及已有數倉上的存量數據等。數據湖能夠將這些不同來源、不同格式的數據集中存儲管理在高性價比的存儲如 OSS 等對象存儲中，并對外提供統一的數據目錄，支持多種計算分析方式，有效解決了企業中面臨的數據孤島問題，同時大大降低了企業存儲和使用數據的成本。

數據湖架構及關鍵技術

企業級數據湖架構如下：

數據湖存儲與格式

數據湖存儲主要以云上對象存儲作為主要介質，其具有低成本、高穩定性、高可擴展性等優點。

數據湖上我們可以采用支持 ACID 的數據湖存儲格式，如 Delta Lake、Hudi、Iceberg。這些數據湖格式有自己的數據 meta 管理能力，能夠支持 Update、Delete 等操作，以批流一體的方式解決了大數據場景下數據實時更新的問題。在當前方案中，我們主要介紹Delta Lake的核心能力和應用場景。

Delta Lake 的核心能力

Delta Lake 是一個統一的數據管理系統，為云上數據湖帶來數據可靠性和快速分析。Delta Lake 運行在現有數據湖之上，并且與 Apache Spark 的 API 完全兼容。使用Delta Lake，您可以加快高質量數據導入數據湖的速度，團隊也可以在云服務上快速使用這些數據，安全且可擴展。

ACID 事務性：Delta Lake 在多個寫操作之間提供 ACID 事務性。每一次寫操作都是一個事務操作，事務日志（Transaction Log）中記錄的寫操作都有一個順序序列。事務日志（Transaction Log）跟蹤了文件級別的寫操作，并使用了樂觀鎖進行并發控制，這非常適用于數據湖，因為嘗試修改相同文件的多次寫操作的情況并不經常發生。當發生沖突時，Delta Lake 會拋出一個并發修改異常，拋給供用戶處理并重試其作業。Delta Lake 還提供了最高級別的隔離（可序列化隔離），允許工程師不斷地向目錄或表寫入數據，而使用者不斷地從同一目錄或表讀取數據，讀取數據時會看到數據的最新快照。

Schema 管理（Schema management）：Delta Lake 會自動驗證正在寫入的DataFrame 的 Schema 是否與表的 Schema 兼容。若表中存在但 DataFrame 中不存在的列則會被設置為 null。如果 DataFrame 中有額外的列不在表中，那么該操作將會拋出異常。Delta Lake 具有 DDL（數據定義語言）顯式添加新列的功能，并且能夠自動更新 Schema。

可伸縮的元數據（Metadata）處理：Delta Lake 將表或目錄的元數據信息存儲在事務日志（Transaction Log）中，而不是元數據 Metastore 中。這使得 Delta Lake夠在固定時間內列出大目錄中的文件，并且在讀取數據時效率很高。

數據版本控制和時間旅行（Time Travel）：Delta Lake 允許用戶讀取表或目錄的歷史版本快照。當文件在寫入過程中被修改時，Delta Lake 會創建文件的新的版本并保留舊版本。當用戶想要讀取表或目錄的較舊版本時，他們可以向 Apach Spark的 read API 提供時間戳或版本號，Delta Lake 根據事務日志（Transaction Log）中的信息來構建該時間戳或版本的完整快照。這非常方便用戶來復現實驗和報告，如果需要，還可以將表還原為舊版本。
統一批流一體：除了批處理寫入之外，Delta Lake 還可以作為 Apache Spark 的結構化流的高效流接收器（Streaming Sink）。與 ACID 事務和可伸縮元數據處理相結合，高效的流接收器（Streaming Sink）支持大量近實時的分析用例，而無需維護復雜的流和批處理管道。

記錄更新和刪除：Delta Lake 將支持合并、更新和刪除的 DML（數據管理語言）命令。這使得工程師可以輕松地在數據湖中插入和刪除記錄，并簡化他們的變更數據捕獲和 GDPR（一般數據保護條例）用例。由于 Delta Lake 在文件級粒度上進行跟蹤和修改數據，因此它比讀取和覆蓋整個分區或表要高效得多。

數據湖構建與管理

1. 數據入湖

企業的原始數據存在于多種數據庫或存儲系統，如關系數據庫 MySQL、日志系統SLS、NoSQL 存儲 HBase、消息數據庫 Kafka 等。其中大部分的在線存儲都面向在線事務型業務，并不適合在線分析的場景，所以需要將數據以無侵入的方式同步至成本更低且更適合計算分析的對象存儲。

常用的數據同步方式有基于 DataX、Sqoop 等數據同步工具做批量同步；同時在對于實時性要求較高的場景下，配合使用 Kafka+spark Streaming / flink 等流式同步鏈路。目前很多云廠商提供了一站式入湖的解決方案，幫助客戶以更快捷更低成本的方式實現數據入湖，如阿里云 DLF 數據入湖。

2. 統一元數據服務

對象存儲本身是沒有面向大數據分析的語義的，需要結合 Hive Metastore Service 等元數據服務為上層各種分析引擎提供數據的 Meta 信息。數據湖元數據服務的設計目標是能夠在大數據引擎、存儲多樣性的環境下，構建不同存儲系統、格式和不同計算引擎統一元數據視圖，并具備統一的權限、元數據，且需要兼容和擴展開源大數據生態元數據服務，支持自動獲取元數據，并達到一次管理多次使用的目的，這樣既能夠兼容開源生態，也具備極大的易用性。

數據湖計算與分析

相比于數據倉庫，數據湖以更開放的方式對接多種不同的計算引擎，如傳統開源大數據計算引擎 Hive、Spark、Presto、Flink 等，同時也支持云廠商自研的大數據引擎，如阿里云 MaxCompute、Hologres 等。在數據湖存儲與計算引擎之間，一般還會提供數據湖加速的服務，以提高計算分析的性能，同時減少帶寬的成本和壓力。

Databricks 數據洞察-商業版的 Spark 數據計算與分析引擎

DataBricks 數據洞察（DDI）做為阿里云上全托管的 Spark 分析引擎，能夠簡單快速幫助用戶對數據湖的數據進行計算與分析。

Saas 全托管 Spark：免運維，無需關注底層資源情況，降低運維成本，聚焦分析業務
完整 Spark 技術棧集成：一站式集成 Spark 引擎和 Delta Lake 數據湖，100%兼容開源 Spark 社區版；Databricks 做商業支持，最快體驗 Spark 最新版本特性
總成本降低：商業版本 Spark 及 Delta Lake 性能優勢顯著；同時基于計算存儲分離架構，存儲依托阿里云 OSS 對象存儲，借助阿里云 JindoFS 緩存層加速；能夠有效降低集群總體使用成本
高品質支持以及 SLA 保障：阿里云和 Databricks 提供覆蓋 Spark 全棧的技術支持；提供商業化 SLA 保障與7*24小時 Databricks 專家支持服務

Databricks 數據洞察+ DLF 數據湖構建與流批一體分析實踐

企業構建和應用數據湖一般需要經歷數據入湖、數據湖存儲與管理、數據湖探索與分析等幾個過程。本文主要介紹基于阿里云數據湖構建（DLF）+Databricks 數據洞察（DDI）構建一站式的數據入湖，批流一體數據分析實戰。

流處理場景：

實時場景維護更新兩張 Delta 表：

delta_aggregates_func 表：RDS 數據實時入湖 。
delta_aggregates_metrics 表：工業 metric 數據通過 IoT 平臺采集到云 Kafka ，經由 Spark Structured Streaming 實時入湖。

批處理場景：

以實時場景生成兩張 Delta 作為數據源，進行數據分析執行 Spark jobs，通過 Databrick 數據洞察作業調度定時執行。

前置條件

1. 服務開通

確保 DLF、OSS、Kafka、DDI、RDS、DTS 等云產品服務已開通。注意 DLF、RDS、Kafka、DDI 實例均需在同一 Region 下。

2. RDS 數據準備

RDS 數據準備，在 RDS 中創建數據庫 dlfdb。在賬戶中心創建能夠讀取 engine_funcs數據庫的用戶賬號，如 dlf_admin。

通過 DMS 登錄數據庫，運行一下語句創建 engine_funcs 表，及插入少量數據。

CREATE TABLE `engine_funcs` (  `emp_no` int(11) NOT NULL,  `engine_serial_number` varchar(20) NOT NULL,  `engine_serial_name` varchar(20) NOT NULL,  `target_engine_serial_number` varchar(20) NOT NULL,  `target_engine_serial_name` varchar(20) NOT NULL,  `operator` varchar(16) NOT NULL,  `create_time` DATETIME NOT NULL,  `update_time` DATETIME NOT NULL,  PRIMARY KEY (`emp_no`)) ENGINE=InnoDB DEFAULT CHARSET=utf8INSERT INTO `engine_funcs` VALUES (10001,'1107108133','temperature','1107108144','temperature','/', now(), now());INSERT INTO `engine_funcs` VALUES (10002,'1107108155','temperature','1107108133','temperature','/', now(), now());INSERT INTO `engine_funcs` VALUES (10003,'1107108155','runTime','1107108166','speed','/', now(), now());INSERT INTO `engine_funcs` VALUES (10004,'1107108177','pressure','1107108155','electricity','/', now(), now());INSERT INTO `engine_funcs` VALUES (10005,'1107108188','flow' ,'1107108111','runTime','/', now(), now()); 

RDS數據實時入湖

1. 創建數據源

進入 DLF 控制臺界面：https://dlf.console.aliyun.com/cn-hangzhou/home，點擊菜單 數據入湖 -> 數據源管理。
點擊 新建數據源。填寫連接名稱，選擇數據準備中的使用的 RDS 實例，填寫賬號密碼，點擊“連接測試”驗證網絡連通性及賬號可用性。

點擊下一步，確定，完成數據源創建。

2. 創建元數據庫

在 OSS 中新建 Bucket，databricks-data-source；

點擊左側菜單“元數據管理”->“元數據庫”，點擊“新建元數據庫”。填寫名稱，新建目錄 dlf/，并選擇。

3. 創建入湖任務

點擊菜單“數據入湖”->“入湖任務管理”，點擊“新建入湖任務”。
選擇“關系數據庫實時入湖”，按照下圖的信息填寫數據源、目標數據湖、任務配置等信息。并保存。
配置數據源，選擇剛才新建的“dlf”連接，使用表路徑 “dlf/engine_funcs”，選擇新建 dts 訂閱，填寫名稱。

回到任務管理頁面，點擊“運行”新建的入湖任務。就會看到任務進入“初始化中”狀態，隨后會進入“運行”狀態。
點擊“詳情”進入任務詳情頁，可以看到相應的數據庫表信息。

該數據入湖任務，屬于全量+增量入湖，大約3至5分鐘后，全量數據會完成導入，隨后自動進入實時監聽狀態。如果有數據更新，則會自動更新至 Delta Lake 數據中。

數據湖探索與分析

DLF 數據查詢探索

DLF 產品提供了輕量級的數據預覽和探索功能，點擊菜單“數據探索”->“SQL 查詢”進入數據查詢頁面。

在元數據庫表中，找到“fjl_dlf”，展開后可以看到 engine_funcs_delta 表已經自動創建完成。雙擊該表名稱，右側 sql 編輯框會出現查詢該表的 sql 語句，點擊“運行”，即可獲得數據查詢結果。

回到 DMS 控制臺，運行下方 DELETE 和 INSERT SQL 語句。

DELETE FROM `engine_funcs` where `emp_no` = 10001;UPDATE `engine_funcs` SET `operator` = '+', `update_time` = NOW() WHERE `emp_no` =10002;INSERT INTO `engine_funcs` VALUES (20001,'1107108199','speed','1107108122','runTime','*', now(), now()); 

大約1至3分鐘后，在 DLF 數據探索再次執行剛才的 select 語句，所有的數據更新已經同步至數據湖中。

創建 Databricks 數據洞察（DDI）集群

集群創建完成后，點擊“詳情”進入詳情頁，添加當前訪問機器 ip 白名單。

點擊 Notebook 進入交互式分析頁查詢同步至 Delta Lake 中 engine_funcs_delta 表數據。

IoT 平臺采集到云 Kafka 數據實時寫入 Delta Lake

1.引入 spark-sql-kafka 三方依賴

%spark.confspark.jars.packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1 

2.使用 UDF 函數定義流數據寫入 Delta Lake 的 Merge 規則

發往 Kafka 的測試數據的格式：

{"sn": "1107108111","temperature": "12" ,"speed":"1115", "runTime":"160","pressure":"210","electricity":"380","flow":"740","dia":"330"}{"sn": "1107108122","temperature": "13" ,"speed":"1015", "runTime":"150","pressure":"220","electricity":"390","flow":"787","dia":"340"}{"sn": "1107108133","temperature": "14" ,"speed":"1215", "runTime":"140","pressure":"230","electricity":"377","flow":"777","dia":"345"}{"sn": "1107108144","temperature": "15" ,"speed":"1315", "runTime":"145","pressure":"240","electricity":"367","flow":"730","dia":"430"}{"sn": "1107108155","temperature": "16" ,"speed":"1415", "runTime":"155","pressure":"250","electricity":"383","flow":"750","dia":"345"}{"sn": "1107108166","temperature": "10" ,"speed":"1515", "runTime":"145","pressure":"260","electricity":"350","flow":"734","dia":"365"}{"sn": "1107108177","temperature": "12" ,"speed":"1115", "runTime":"160","pressure":"210","electricity":"377","flow":"733","dia":"330"}{"sn": "1107108188","temperature": "13" ,"speed":"1015", "runTime":"150","pressure":"220","electricity":"381","flow":"737","dia":"340"}{"sn": "1107108199","temperature": "14" ,"speed":"1215", "runTime":"140","pressure":"230","electricity":"378","flow":"747","dia":"345"} 

%sparkimport org.apache.spark.sql._import io.delta.tables._def upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) { microBatchOutputDF.createOrReplaceTempView("dataStream") // 對流數據DF執行列轉行的操作； val df=microBatchOutputDF.sparkSession.sql(s""" select `sn`, stack(7, 'temperature', `temperature`, 'speed', `speed`, 'runTime', `runTime`, 'pressure', `pressure`, 'electricity', `electricity`, 'flow', `flow` , 'dia', `dia`) as (`name`, `value` ) from dataStream """) df.createOrReplaceTempView("updates") // 實現實時更新動態的數據，結果merge到表里面 val mergedf=df.sparkSession.sql(s""" MERGE INTO delta_aggregates_metrics t USING updates s ON s.sn = t.sn and s.name=t.name WHEN MATCHED THEN UPDATE SET t.value = s.value, t.update_time=current_timestamp() WHEN NOT MATCHED THEN INSERT (t.sn,t.name,t.value ,t.create_time,t.update_time) values (s.sn,s.name,s.value,current_timestamp(),current_timestamp()) """)} 

3.使用 Spark Structured Streaming 實時流寫入 Delta Lake

%sparkimport org.apache.spark.sql.functions._import org.apache.spark.sql.streaming.Triggerdef getquery(checkpoint_dir:String,servers:String,topic:String ){    var streamingInputDF =    spark.readStream    .format("kafka")    .option("kafka.bootstrap.servers", servers)    .option("subscribe", topic)         .option("startingOffsets", "latest")      .option("minPartitions", "10")      .option("failOnDataLoss", "true")    .load()var streamingSelectDF =   streamingInputDF   .select(    get_json_object(($"value").cast("string"), "$.sn").alias("sn"),   get_json_object(($"value").cast("string"), "$.temperature").alias("temperature"),   get_json_object(($"value").cast("string"), "$.speed").alias("speed"),   get_json_object(($"value").cast("string"), "$.runTime").alias("runTime"),   get_json_object(($"value").cast("string"), "$.electricity").alias("electricity"),   get_json_object(($"value").cast("string"), "$.flow").alias("flow"),   get_json_object(($"value").cast("string"), "$.dia").alias("dia"),   get_json_object(($"value").cast("string"), "$.pressure").alias("pressure")   )val query = streamingSelectDF      .writeStream      .format("delta")      .option("checkpointLocation", checkpoint_dir)      .trigger(Trigger.ProcessingTime("5 seconds")) // 執行流處理時間間隔      .foreachBatch(upsertToDelta _) //引用upsertToDelta函數      .outputMode("update")      .start()} 

4. 執行程序

%sparkval my_checkpoint_dir="oss://databricks-data-source/checkpoint/ck"val servers= "***.***.***.***:9092"val topic= "your-topic"getquery(my_checkpoint_dir,servers,topic) 

5. 啟動 Kafka 并向生產里發送測試數據

查詢數據實時寫入并更新

查詢從 MySQL 實時同步入湖的 engine_funcs_delta 數據

%sparkval rds_dataV=spark.table("fjl_dlf.engine_funcs_delta")rds_dataV.show() 

批處理作業

結合業務，需要將對應的 delta_aggregates_metrics 里的 Value 參數 join 到engine_funcs_delta 表里

%spark//讀取實時更新的delta_aggregates_metrics數據表val aggregateDF=spark.table("log_data_warehouse_dlf.delta_aggregates_metrics")//讀取實時更新的engine_funcs_delta函數表val rds_dataV=spark.table("fjl_dlf.engine_funcs_delta").drop("create_time","update_time")// rds_dataV.show()val aggregateSDF= aggregateDF.withColumnRenamed("value","esn_value").withColumnRenamed("name","engine_serial_name").withColumnRenamed("sn","engine_serial_number")// aggregateSDF.show()val aggregateTDF=aggregateDF.withColumnRenamed("value","tesn_value").withColumnRenamed("name","target_engine_serial_name").withColumnRenamed("sn","target_engine_serial_number").drop("create_time","update_time")// aggregateTDF.show()//將對應的delta_aggregates_metrics里的Value參數 join到engine_funcs_delta表里；val  resdf=rds_dataV.join(aggregateSDF,Seq("engine_serial_name","engine_serial_number"),"left").join(aggregateTDF,Seq("target_engine_serial_number","target_engine_serial_name"),"left")            .selectExpr("engine_serial_number","engine_serial_name","esn_value","target_engine_serial_number","target_engine_serial_name","tesn_value","operator","create_time","update_time")//數據展示resdf.show(false)// 將結果寫入到Delta表里面resdf.write.format("delta")    .mode("append")    .saveAsTable("log_data_warehouse_dlf.delta_result") 

性能優化：OPTIMIZE & Z-Ordering

在流處理場景下會產生大量的小文件，大量小文件的存在會嚴重影響數據系統的讀性能。Delta Lake 提供了 OPTIMIZE 命令，可以將小文件進行合并壓縮，另外，針對 Ad-Hoc 查詢場景，由于涉及對單表多個維度數據的查詢，我們借助 Delta Lake 提供的 Z-Ordering 機制，可以有效提升查詢的性能。從而極大提升讀取表的性能。DeltaLake 本身提供了 Auto Optimize 選項，但是會犧牲少量寫性能，增加數據寫入 delta 表的延遲。相反，執行 OPTIMIZE 命令并不會影響寫的性能，因為 Delta Lake 本身支持 MVCC，支持 OPTIMIZE 的同時并發執行寫操作。因此，我們采用定期觸發執行 OPTIMIZE 的方案，每小時通過 OPTIMIZE 做一次合并小文件操作，同時執行 VACCUM 來清理過期數據文件:

OPTIMIZE log_data_warehouse_dlf.delta_result ZORDER by engine_serial_number;VACUUM log_data_warehouse_dlf.delta_result RETAIN 1 HOURS;