簡介：本文介紹了Databricks企業(yè)版Delta Lake的性能優(yōu)勢，借助這些特性能夠大幅提升Spark SQL的查詢性能，加快Delta表的查詢速度。

作者：

李錦桂（錦犀） 阿里云開源大數(shù)據(jù)平臺開發(fā)工程師

王曉龍（筱龍） 阿里云開源大數(shù)據(jù)平臺技術(shù)專家

背景介紹

Databricks是全球領(lǐng)先的Data+AI企業(yè)，是Apache Spark的創(chuàng)始公司，也是Spark的最大代碼貢獻(xiàn)者，核心圍繞Spark、Delta Lake、MLFlow等開源生態(tài)打造企業(yè)級Lakehouse產(chǎn)品。2020年，Databricks 和阿里云聯(lián)手打造了基于Apache Spark的云上全托管大數(shù)據(jù)分析&AI平臺——Databricks數(shù)據(jù)洞察(DDI，Databricks DataInsight)，為用戶提供數(shù)據(jù)分析、數(shù)據(jù)工程、數(shù)據(jù)科學(xué)和人工智能等方面的服務(wù)，構(gòu)建一體化的Lakehouse架構(gòu)。

Delta Lake是Databricks從2016年開始在內(nèi)部研發(fā)的一款支持事務(wù)的數(shù)據(jù)湖產(chǎn)品，于2019年正式開源。除了社區(qū)主導(dǎo)的開源版Delta Lake OSS，Databricks商業(yè)產(chǎn)品里也提供了企業(yè)版Spark&Detla Lake引擎，本文將介紹企業(yè)版提供的產(chǎn)品特性如何優(yōu)化性能，助力高效訪問Lakehouse。

針對小文件問題的優(yōu)化解法

在Delta Lake中頻繁執(zhí)行merge, update, insert操作，或者在流處理場景下不斷往Delta表中插入數(shù)據(jù)，會(huì)導(dǎo)致Delta表中產(chǎn)生大量的小文件。小文件數(shù)量的增加一方面會(huì)使得Spark每次串行讀取的數(shù)據(jù)量變少，降低讀取效率，另一方面，使得Delta表的元數(shù)據(jù)增加，元數(shù)據(jù)獲取變慢，從另一個(gè)維度降低表的讀取效率。

為了解決小文件問題，Databricks提供了三個(gè)優(yōu)化特性，從避免小文件的產(chǎn)生和自動(dòng)/手動(dòng)合并小文件兩個(gè)維度來解決Delta Lake的小文件問題。

特性1：優(yōu)化Delta表的寫入，避免小文件產(chǎn)生

在開源版Spark中，每個(gè)executor向partition中寫入數(shù)據(jù)時(shí)，都會(huì)創(chuàng)建一個(gè)表文件進(jìn)行寫入，最終會(huì)導(dǎo)致一個(gè)partition中產(chǎn)生很多的小文件。Databricks對Delta表的寫入過程進(jìn)行了優(yōu)化，對每個(gè)partition，使用一個(gè)專門的executor合并其他executor對該partition的寫入，從而避免了小文件的產(chǎn)生。

該特性由表屬性delta.autoOptimize.optimizeWrite來控制：

• 可以在創(chuàng)建表時(shí)指定

CREATE TABLE student (id INT, name STRING) TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true);

• 也可以修改表屬性

ALTER TABLE table_name SET TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true);

該特性有兩個(gè)優(yōu)點(diǎn)：

通過減少被寫入的表文件數(shù)量，提高寫數(shù)據(jù)的吞吐量；

避免小文件的產(chǎn)生，提升查詢性能。

其缺點(diǎn)也是顯而易見的，由于使用了一個(gè)executor來合并表文件的寫入，從而降低了表文件寫入的并行度，此外，多引入的一層executor需要對寫入的數(shù)據(jù)進(jìn)行shuffle，帶來額外的開銷。因此，在使用該特性時(shí)，需要對場景進(jìn)行評估：

• 該特性適用的場景：頻繁使用MERGE，UPDATE，DELETE，INSERT INTO，CREATE TABLE AS SELECT等SQL語句的場景；
• 該特性不適用的場景：寫入TB級以上數(shù)據(jù)。

特性2：自動(dòng)合并小文件

在流處理場景中，比如流式數(shù)據(jù)入湖場景下，需要持續(xù)的將到達(dá)的數(shù)據(jù)插入到Delta表中，每次插入都會(huì)創(chuàng)建一個(gè)新的表文件用于存儲新到達(dá)的數(shù)據(jù)，假設(shè)每10s觸發(fā)一次，那么這樣的流處理作業(yè)一天產(chǎn)生的表文件數(shù)量將達(dá)到8640個(gè)，且由于流處理作業(yè)通常是long-running的，運(yùn)行該流處理作業(yè)100天將產(chǎn)生上百萬個(gè)表文件。這樣的Delta表，僅元數(shù)據(jù)的維護(hù)就是一個(gè)很大的挑戰(zhàn)，查詢性能更是急劇惡化。

為了解決上述問題，Databricks提供了小文件自動(dòng)合并功能，在每次向Delta表中寫入數(shù)據(jù)之后，會(huì)檢查Delta表中的表文件數(shù)量，如果Delta表中的小文件（size < 128MB的視為小文件）數(shù)量達(dá)到閾值，則會(huì)執(zhí)行一次小文件合并，將Delta表中的小文件合并為一個(gè)新的大文件。

該特性由表屬性delta.autoOptimize.autoCompact控制，和特性delta.autoOptimize.optimizeWrite相同，可以在創(chuàng)建表時(shí)指定，也可以對已創(chuàng)建的表進(jìn)行修改。自動(dòng)合并的閾值由spark.databricks.delta.autoCompact.minNumFiles控制，默認(rèn)為50，即小文件數(shù)量達(dá)到50會(huì)執(zhí)行表文件合并；合并后產(chǎn)生的文件最大為128MB，如果需要調(diào)整合并后的目標(biāo)文件大小，可以通過調(diào)整配置spark.databricks.delta.autoCompact.maxFileSize實(shí)現(xiàn)。

特性3：手動(dòng)合并小文件

自動(dòng)小文件合并會(huì)在對Delta表進(jìn)行寫入，且寫入后表中小文件達(dá)到閾值時(shí)被觸發(fā)。除了自動(dòng)合并之外，Databricks還提供了Optimize命令使用戶可以手動(dòng)合并小文件，優(yōu)化表結(jié)構(gòu)，使得表文件的結(jié)構(gòu)更加緊湊。在實(shí)現(xiàn)上Optimize使用bin-packing算法，該算法不但會(huì)合并表中的小文件，且合并后生成的表文件也更均衡（表文件大小相近）。例如，我們要對Delta表student的表文件進(jìn)行優(yōu)化，僅需執(zhí)行如下命令即可實(shí)現(xiàn)：

OPTIMIZE student;

Optimize命令不但支持全表小文件的合并，還支持特定的分區(qū)的表文件的合并，例如，我們可以僅對date大于2017-01-01的分區(qū)中的小文件進(jìn)行合并：

OPTIMIZE student WHERE date >= '2017-01-01'

從Databricks數(shù)據(jù)洞察產(chǎn)品上的試驗(yàn)數(shù)據(jù)看，Optimize能使查詢性能達(dá)到8x以上的提升。

媲美企業(yè)級數(shù)據(jù)庫的查詢優(yōu)化技術(shù)

Databricks在數(shù)據(jù)查詢方面也做了諸多優(yōu)化，包括：

特性1：Data Skipping

在數(shù)據(jù)查詢系統(tǒng)中，有兩種經(jīng)典的查詢優(yōu)化 技術(shù)：一種是以更快的速度處理數(shù)據(jù)，另一種是通過跳過不相關(guān)的數(shù)據(jù)，減少需要掃描的數(shù)據(jù)量。Data Skipping屬于后一種優(yōu)化技術(shù)，通過表文件的統(tǒng)計(jì)信息跳過不相關(guān)的表文件，從而提升查詢性能。

在向Delta表中新增表文件時(shí)，Delta Lake會(huì)在Delta表的元數(shù)據(jù)中存儲該表文件中的數(shù)據(jù)前32列的統(tǒng)計(jì)信息，包括數(shù)據(jù)列的最大最小值，以及為null的行的數(shù)量，在查詢時(shí)，Databricks會(huì)利用這些統(tǒng)計(jì)信息提升查詢性能。例如：對于一張Delta表的x列，假設(shè)該表的一個(gè)表文件x列的最小值為5，最大值為10，如果查詢條件為 where x < 3，則根據(jù)表文件的統(tǒng)計(jì)信息，我們可以得出結(jié)論：該表文件中一定不包含我們需要的數(shù)據(jù)，因此我們可以直接跳過該表文件，減少掃描的數(shù)據(jù)量，進(jìn)而提升查詢性能。

Data Skipping的實(shí)現(xiàn)原理和布隆過濾器類似，通過查詢條件判斷表文件中是否可能存在需要查詢的數(shù)據(jù)，從而減少需要掃描的數(shù)據(jù)量。如果表文件不可能存在查詢的數(shù)據(jù)，則可以直接跳過，如果表文件可能存在被查詢的數(shù)據(jù)，則需要掃描表文件。

為了能盡可能多的跳過和查詢無關(guān)的表文件，我們需要縮小表文件的min-max的差距，使得相近的數(shù)據(jù)盡可能在文件中聚集。舉一個(gè)簡單的例子，假設(shè)一張表包含10個(gè)表文件，對于表中的x列，它的取值為[1, 10]，如果每個(gè)表文件的x列的分布均為[1, 10]，則對于查詢條件：where x < 3，無法跳過任何一個(gè)表文件，因此，也無法實(shí)現(xiàn)性能提升，而如果每個(gè)表文件的min-max均為0，即在表文件1的x列分布為[1, 1]，表文件2的x列分布為[2, 2]...，則對于查詢條件：where x < 3，可以跳過80%的表文件。受該思想的啟發(fā)，Databricks支持使用Z-Ordering來對數(shù)據(jù)進(jìn)行聚集，縮小表文件的min-max差距，提升查詢性能。下面我們介紹Z-Ordering優(yōu)化的原理和使用。

特性2：Z-Ordering優(yōu)化

如上一節(jié)所解釋的，為了能盡可能多的跳過無關(guān)的表文件，表文件中作為查詢條件的列應(yīng)該盡可能緊湊（即min-max的差距盡可能小）。Z-Ordering就可以實(shí)現(xiàn)該功能，它可以在多個(gè)維度上將關(guān)聯(lián)的信息存儲到同一組文件中，因此確切來說，Z-Ordering實(shí)際是一種數(shù)據(jù)布局優(yōu)化算法，但結(jié)合Data Skipping，它可以顯著提升查詢性能。

Z-Ordering的使用非常簡單，對于表events，如果經(jīng)常使用列eventType和generateTime作為查詢條件，那么執(zhí)行命令：

OPTIMIZE events ZORDER BY (eventType, generateTime)

Delta表會(huì)使用列eventType和generateTime調(diào)整數(shù)據(jù)布局，使得表文件中eventType和generateTime盡可能緊湊。

根據(jù)我們在Databricks DataInsight上的試驗(yàn)，使用Z-Ordering優(yōu)化能達(dá)到40倍的性能提升，具體的試驗(yàn)案例參考文末Databricks數(shù)據(jù)洞察的官方文檔。

特性3：布隆過濾器索引

布隆過濾器也是一項(xiàng)非常有用的Data-skipping技術(shù)。該技術(shù)可以快速判斷表文件中是否包含要查詢的數(shù)據(jù)，如果不包含就及時(shí)跳過該文件，從而減少掃描的數(shù)據(jù)量，提升查詢性能。

如果在表的某列上創(chuàng)建了布隆過濾器索引，并且使用where col = "something"作為查詢條件，那么在掃描表中文件時(shí)，我們可以使用布隆過濾器索引得出兩種結(jié)論：文件中肯定不包含col = "something"的行，或者文件有可能包含col = "something"的行。

• 當(dāng)?shù)贸鑫募锌隙ú话琧ol = "something"的行的結(jié)論時(shí)，就可以跳過該文件，從而減少掃描的數(shù)據(jù)量，提升查詢性能。
• 當(dāng)?shù)贸鑫募锌赡馨琧ol = "something"的行的結(jié)論時(shí)，引擎才會(huì)去處理該文件。注意，這里僅僅是判斷該文件中可能包含目標(biāo)數(shù)據(jù)。布隆過濾器定義了一個(gè)指標(biāo)，用于描述出現(xiàn)判斷失誤的概率，即判斷文件中包含需要查詢的數(shù)據(jù)，而實(shí)際上該文件并不包含目標(biāo)數(shù)據(jù)的概率，并稱之為FPP（False Positive Probability: 假陽性概率）。

Databricks支持文件級Bloom過濾器，如果在表的某些列創(chuàng)建了布隆過濾器索引，那么該表的每個(gè)表文件都會(huì)關(guān)聯(lián)一個(gè) Bloom 篩選器索引文件，索引文件存儲在表文件同級目錄下的 _delta_index 子目錄中。在讀取表文文件之前，Databricks會(huì)檢查索引文件，根據(jù)上面的步驟判斷表文件中是否包含需要查詢的數(shù)據(jù)，如果不包含則直接跳過，否則再進(jìn)行處理。

布隆過濾器索引的創(chuàng)建和傳統(tǒng)數(shù)據(jù)庫索引的創(chuàng)建類似，但需要指定假陽性概率和該列可能出現(xiàn)的值的數(shù)量：

CREATE BLOOMFILTER INDEX ON TABLE table_name FOR COLUMNS(col_name OPTIONS (fpp=0.1, numItems=50000000))

根據(jù)我們在Databricks DataInsight上的試驗(yàn)，使用布隆過濾器索引能達(dá)到3倍以上的性能提升，試驗(yàn)案例參考文末Databricks數(shù)據(jù)洞察的官方文檔。

特性4：動(dòng)態(tài)文件剪枝

動(dòng)態(tài)文件剪枝（Dynamic File Pruning, DFP）和動(dòng)態(tài)分區(qū)剪枝（Dynamic Partition Pruning）相似，都是在維表和事實(shí)表的Join執(zhí)行階段進(jìn)行剪枝，減少掃描的數(shù)據(jù)量，提升查詢效率。
下面我們以一個(gè)簡單的查詢?yōu)槔齺斫榻BDFP的原理：

SELECT sum(ss_quantity) FROM store_sales JOIN item ON ss_item_sk = i_item_sk WHERE i_item_id = 'AAAAAAAAICAAAAAA'

在該查詢中，item為維表（數(shù)據(jù)量很少），store_sales為事實(shí)表（數(shù)據(jù)量非常大），where查詢條件作用于維表上。如果不開啟DFP，那么該查詢的邏輯執(zhí)行計(jì)劃如下：

從上圖可以看出，先對store_sales進(jìn)行全表掃描，然后再和過濾后的item表的行進(jìn)行join，雖然結(jié)果僅有4萬多條數(shù)據(jù)，但卻掃描了表store_sales中的80多億條數(shù)據(jù)。針對該查詢，很直觀的優(yōu)化是：先查詢出表item中i_item_id = 'AAAAAAAAICAAAAAA'數(shù)據(jù)行，然后將這些數(shù)據(jù)行的i_item_sk值作為表store_sales的ss_item_sk的查詢條件在表store_sales的SCAN階段進(jìn)行過濾，結(jié)合我們在上面介紹的Data Skipping技術(shù)，可以大幅減少表文件的掃描。這一思路正是DFP的根本原理，啟動(dòng)DFP后的邏輯執(zhí)行計(jì)劃如下圖所示：

可以看到，在啟用DFP之后，過濾條件被下推到SCAN操作中，僅掃描了600多萬條store_sales中的數(shù)據(jù)，從結(jié)果上看，啟動(dòng)DFP后，該條查詢實(shí)現(xiàn)了10倍的性能提升，此外，Databricks還針對該特性對TPC-DS測試，測試發(fā)現(xiàn)啟用DFP后，TPC-DS的第15條查詢達(dá)到了8倍的性能提升，有36條查詢實(shí)現(xiàn)了2倍及以上的性能提升。

總結(jié)

前文概括介紹了Databricks企業(yè)版Delta Lake的性能優(yōu)勢，借助這些特性能夠大幅提升Spark SQL的查詢性能，加快Delta表的查詢速度。原文鏈接本文為阿里云原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。?

總結(jié)

以上是生活随笔為你收集整理的Databricks 企业版 SparkDelta Lake 引擎助力 Lakehouse 高效访问的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。