背景

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

數(shù)據(jù)湖作為一個(gè)集中化的數(shù)據(jù)存儲倉庫，支持的數(shù)據(jù)類型具有多樣性，包括結(jié)構(gòu)化、半結(jié)構(gòu)化以及非結(jié)構(gòu)化的數(shù)據(jù)，數(shù)據(jù)來源上包含數(shù)據(jù)庫數(shù)據(jù)、binglog 增量數(shù)據(jù)、日志數(shù)據(jù)以及已有數(shù)倉上的存量數(shù)據(jù)等。數(shù)據(jù)湖能夠?qū)⑦@些不同來源、不同格式的數(shù)據(jù)集中存儲管理在高性價(jià)比的存儲如 OSS 等對象存儲中，并對外提供統(tǒng)一的數(shù)據(jù)目錄，支持多種計(jì)算分析方式，有效解決了企業(yè)中面臨的數(shù)據(jù)孤島問題，同時(shí)大大降低了企業(yè)存儲和使用數(shù)據(jù)的成本。

數(shù)據(jù)湖架構(gòu)及關(guān)鍵技術(shù)

企業(yè)級數(shù)據(jù)湖架構(gòu)如下：

數(shù)據(jù)湖存儲與格式

數(shù)據(jù)湖存儲主要以云上對象存儲作為主要介質(zhì)，其具有低成本、高穩(wěn)定性、高可擴(kuò)展性等優(yōu)點(diǎn)。

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

Delta Lake 的核心能力

Delta Lake 是一個(gè)統(tǒng)一的數(shù)據(jù)管理系統(tǒng)，為云上數(shù)據(jù)湖帶來數(shù)據(jù)可靠性和快速分析。Delta Lake 運(yùn)行在現(xiàn)有數(shù)據(jù)湖之上，并且與 Apache Spark 的 API 完全兼容。使用Delta Lake，您可以加快高質(zhì)量數(shù)據(jù)導(dǎo)入數(shù)據(jù)湖的速度，團(tuán)隊(duì)也可以在云服務(wù)上快速使用這些數(shù)據(jù)，安全且可擴(kuò)展。

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

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

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

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

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

數(shù)據(jù)湖構(gòu)建與管理

1. 數(shù)據(jù)入湖

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

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

2. 統(tǒng)一元數(shù)據(jù)服務(wù)

對象存儲本身是沒有面向大數(shù)據(jù)分析的語義的，需要結(jié)合 Hive Metastore Service 等元數(shù)據(jù)服務(wù)為上層各種分析引擎提供數(shù)據(jù)的 Meta 信息。數(shù)據(jù)湖元數(shù)據(jù)服務(wù)的設(shè)計(jì)目標(biāo)是能夠在大數(shù)據(jù)引擎、存儲多樣性的環(huán)境下，構(gòu)建不同存儲系統(tǒng)、格式和不同計(jì)算引擎統(tǒng)一元數(shù)據(jù)視圖，并具備統(tǒng)一的權(quán)限、元數(shù)據(jù)，且需要兼容和擴(kuò)展開源大數(shù)據(jù)生態(tài)元數(shù)據(jù)服務(wù)，支持自動獲取元數(shù)據(jù)，并達(dá)到一次管理多次使用的目的，這樣既能夠兼容開源生態(tài)，也具備極大的易用性。

數(shù)據(jù)湖計(jì)算與分析

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

Databricks 數(shù)據(jù)洞察-商業(yè)版的 Spark 數(shù)據(jù)計(jì)算與分析引擎

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

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

Databricks 數(shù)據(jù)洞察+ DLF 數(shù)據(jù)湖構(gòu)建與流批一體分析實(shí)踐

企業(yè)構(gòu)建和應(yīng)用數(shù)據(jù)湖一般需要經(jīng)歷數(shù)據(jù)入湖、數(shù)據(jù)湖存儲與管理、數(shù)據(jù)湖探索與分析等幾個(gè)過程。本文主要介紹基于阿里云數(shù)據(jù)湖構(gòu)建（DLF）+Databricks 數(shù)據(jù)洞察（DDI）構(gòu)建一站式的數(shù)據(jù)入湖，批流一體數(shù)據(jù)分析實(shí)戰(zhàn)。

流處理場景：

實(shí)時(shí)場景維護(hù)更新兩張 Delta 表：

delta_aggregates_func 表：RDS 數(shù)據(jù)實(shí)時(shí)入湖 。
delta_aggregates_metrics 表：工業(yè) metric 數(shù)據(jù)通過 IoT 平臺采集到云 Kafka ，經(jīng)由 Spark Structured Streaming 實(shí)時(shí)入湖。

批處理場景：

以實(shí)時(shí)場景生成兩張 Delta 作為數(shù)據(jù)源，進(jìn)行數(shù)據(jù)分析執(zhí)行 Spark jobs，通過 Databrick 數(shù)據(jù)洞察作業(yè)調(diào)度定時(shí)執(zhí)行。

前置條件

1. 服務(wù)開通

確保 DLF、OSS、Kafka、DDI、RDS、DTS 等云產(chǎn)品服務(wù)已開通。注意 DLF、RDS、Kafka、DDI 實(shí)例均需在同一 Region 下。

2. RDS 數(shù)據(jù)準(zhǔn)備

RDS 數(shù)據(jù)準(zhǔn)備，在 RDS 中創(chuàng)建數(shù)據(jù)庫 dlfdb。在賬戶中心創(chuàng)建能夠讀取 engine_funcs數(shù)據(jù)庫的用戶賬號，如 dlf_admin。

通過 DMS 登錄數(shù)據(jù)庫，運(yùn)行一下語句創(chuàng)建 engine_funcs 表，及插入少量數(shù)據(jù)。

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數(shù)據(jù)實(shí)時(shí)入湖

1. 創(chuàng)建數(shù)據(jù)源

進(jìn)入 DLF 控制臺界面：https://dlf.console.aliyun.com/cn-hangzhou/home，點(diǎn)擊菜單 數(shù)據(jù)入湖 -> 數(shù)據(jù)源管理。
點(diǎn)擊 新建數(shù)據(jù)源。填寫連接名稱，選擇數(shù)據(jù)準(zhǔn)備中的使用的 RDS 實(shí)例，填寫賬號密碼，點(diǎn)擊“連接測試”驗(yàn)證網(wǎng)絡(luò)連通性及賬號可用性。

點(diǎn)擊下一步，確定，完成數(shù)據(jù)源創(chuàng)建。

2. 創(chuàng)建元數(shù)據(jù)庫

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

點(diǎn)擊左側(cè)菜單“元數(shù)據(jù)管理”->“元數(shù)據(jù)庫”，點(diǎn)擊“新建元數(shù)據(jù)庫”。填寫名稱，新建目錄 dlf/，并選擇。

3. 創(chuàng)建入湖任務(wù)

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

回到任務(wù)管理頁面，點(diǎn)擊“運(yùn)行”新建的入湖任務(wù)。就會看到任務(wù)進(jìn)入“初始化中”狀態(tài)，隨后會進(jìn)入“運(yùn)行”狀態(tài)。
點(diǎn)擊“詳情”進(jìn)入任務(wù)詳情頁，可以看到相應(yīng)的數(shù)據(jù)庫表信息。

該數(shù)據(jù)入湖任務(wù)，屬于全量+增量入湖，大約3至5分鐘后，全量數(shù)據(jù)會完成導(dǎo)入，隨后自動進(jìn)入實(shí)時(shí)監(jiān)聽狀態(tài)。如果有數(shù)據(jù)更新，則會自動更新至 Delta Lake 數(shù)據(jù)中。

數(shù)據(jù)湖探索與分析

DLF 數(shù)據(jù)查詢探索

DLF 產(chǎn)品提供了輕量級的數(shù)據(jù)預(yù)覽和探索功能，點(diǎn)擊菜單“數(shù)據(jù)探索”->“SQL 查詢”進(jìn)入數(shù)據(jù)查詢頁面。

在元數(shù)據(jù)庫表中，找到“fjl_dlf”，展開后可以看到 engine_funcs_delta 表已經(jīng)自動創(chuàng)建完成。雙擊該表名稱，右側(cè) sql 編輯框會出現(xiàn)查詢該表的 sql 語句，點(diǎn)擊“運(yùn)行”，即可獲得數(shù)據(jù)查詢結(jié)果。

回到 DMS 控制臺，運(yùn)行下方 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 數(shù)據(jù)探索再次執(zhí)行剛才的 select 語句，所有的數(shù)據(jù)更新已經(jīng)同步至數(shù)據(jù)湖中。

創(chuàng)建 Databricks 數(shù)據(jù)洞察（DDI）集群

集群創(chuàng)建完成后，點(diǎn)擊“詳情”進(jìn)入詳情頁，添加當(dāng)前訪問機(jī)器 ip 白名單。

點(diǎn)擊 Notebook 進(jìn)入交互式分析頁查詢同步至 Delta Lake 中 engine_funcs_delta 表數(shù)據(jù)。

IoT 平臺采集到云 Kafka 數(shù)據(jù)實(shí)時(shí)寫入 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 函數(shù)定義流數(shù)據(jù)寫入 Delta Lake 的 Merge 規(guī)則

發(fā)往 Kafka 的測試數(shù)據(jù)的格式：

{"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") // 對流數(shù)據(jù)DF執(zhí)行列轉(zhuǎn)行的操作； 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") // 實(shí)現(xiàn)實(shí)時(shí)更新動態(tài)的數(shù)據(jù)，結(jié)果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 實(shí)時(shí)流寫入 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")) // 執(zhí)行流處理時(shí)間間隔      .foreachBatch(upsertToDelta _) //引用upsertToDelta函數(shù)      .outputMode("update")      .start()} 

4. 執(zhí)行程序

%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 并向生產(chǎn)里發(fā)送測試數(shù)據(jù)

查詢數(shù)據(jù)實(shí)時(shí)寫入并更新

查詢從 MySQL 實(shí)時(shí)同步入湖的 engine_funcs_delta 數(shù)據(jù)

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

批處理作業(yè)

結(jié)合業(yè)務(wù)，需要將對應(yīng)的 delta_aggregates_metrics 里的 Value 參數(shù) join 到engine_funcs_delta 表里

%spark//讀取實(shí)時(shí)更新的delta_aggregates_metrics數(shù)據(jù)表val aggregateDF=spark.table("log_data_warehouse_dlf.delta_aggregates_metrics")//讀取實(shí)時(shí)更新的engine_funcs_delta函數(shù)表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()//將對應(yīng)的delta_aggregates_metrics里的Value參數(shù) 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")//數(shù)據(jù)展示resdf.show(false)// 將結(jié)果寫入到Delta表里面resdf.write.format("delta")    .mode("append")    .saveAsTable("log_data_warehouse_dlf.delta_result") 

性能優(yōu)化：OPTIMIZE & Z-Ordering

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

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