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

作者：

李錦桂（錦犀） 阿里云開源大數據平臺開發工程師

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

背景介紹

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

Delta Lake是Databricks從2016年開始在內部研發的一款支持事務的數據湖產品，于2019年正式開源。除了社區主導的開源版Delta Lake OSS，Databricks商業產品里也提供了企業版Spark&Detla Lake引擎，本文將介紹企業版提供的產品特性如何優化性能，助力高效訪問Lakehouse。

針對小文件問題的優化解法

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

為了解決小文件問題，Databricks提供了三個優化特性，從避免小文件的產生和自動/手動合并小文件兩個維度來解決Delta Lake的小文件問題。

特性1：優化Delta表的寫入，避免小文件產生

在開源版Spark中，每個executor向partition中寫入數據時，都會創建一個表文件進行寫入，最終會導致一個partition中產生很多的小文件。Databricks對Delta表的寫入過程進行了優化，對每個partition，使用一個專門的executor合并其他executor對該partition的寫入，從而避免了小文件的產生。

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

• 可以在創建表時指定

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

• 也可以修改表屬性

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

該特性有兩個優點：

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

避免小文件的產生，提升查詢性能。

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

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

特性2：自動合并小文件

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

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

該特性由表屬性delta.autoOptimize.autoCompact控制，和特性delta.autoOptimize.optimizeWrite相同，可以在創建表時指定，也可以對已創建的表進行修改。自動合并的閾值由spark.databricks.delta.autoCompact.minNumFiles控制，默認為50，即小文件數量達到50會執行表文件合并；合并后產生的文件最大為128MB，如果需要調整合并后的目標文件大小，可以通過調整配置spark.databricks.delta.autoCompact.maxFileSize實現。

特性3：手動合并小文件

自動小文件合并會在對Delta表進行寫入，且寫入后表中小文件達到閾值時被觸發。除了自動合并之外，Databricks還提供了Optimize命令使用戶可以手動合并小文件，優化表結構，使得表文件的結構更加緊湊。在實現上Optimize使用bin-packing算法，該算法不但會合并表中的小文件，且合并后生成的表文件也更均衡（表文件大小相近）。例如，我們要對Delta表student的表文件進行優化，僅需執行如下命令即可實現：

OPTIMIZE student;

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

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

從Databricks數據洞察產品上的試驗數據看，Optimize能使查詢性能達到8x以上的提升。

媲美企業級數據庫的查詢優化技術

Databricks在數據查詢方面也做了諸多優化，包括：

特性1：Data Skipping

在數據查詢系統中，有兩種經典的查詢優化 技術：一種是以更快的速度處理數據，另一種是通過跳過不相關的數據，減少需要掃描的數據量。Data Skipping屬于后一種優化技術，通過表文件的統計信息跳過不相關的表文件，從而提升查詢性能。

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

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

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

特性2：Z-Ordering優化

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

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

OPTIMIZE events ZORDER BY (eventType, generateTime)

Delta表會使用列eventType和generateTime調整數據布局，使得表文件中eventType和generateTime盡可能緊湊。

根據我們在Databricks DataInsight上的試驗，使用Z-Ordering優化能達到40倍的性能提升，具體的試驗案例參考文末Databricks數據洞察的官方文檔。

特性3：布隆過濾器索引

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

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

• 當得出文件中肯定不包含col = "something"的行的結論時，就可以跳過該文件，從而減少掃描的數據量，提升查詢性能。
• 當得出文件中可能包含col = "something"的行的結論時，引擎才會去處理該文件。注意，這里僅僅是判斷該文件中可能包含目標數據。布隆過濾器定義了一個指標，用于描述出現判斷失誤的概率，即判斷文件中包含需要查詢的數據，而實際上該文件并不包含目標數據的概率，并稱之為FPP（False Positive Probability: 假陽性概率）。

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

布隆過濾器索引的創建和傳統數據庫索引的創建類似，但需要指定假陽性概率和該列可能出現的值的數量：

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

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

特性4：動態文件剪枝

動態文件剪枝（Dynamic File Pruning, DFP）和動態分區剪枝（Dynamic Partition Pruning）相似，都是在維表和事實表的Join執行階段進行剪枝，減少掃描的數據量，提升查詢效率。
下面我們以一個簡單的查詢為例來介紹DFP的原理：

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

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

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

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

總結

前文概括介紹了Databricks企業版Delta Lake的性能優勢，借助這些特性能夠大幅提升Spark SQL的查詢性能，加快Delta表的查詢速度。原文鏈接本文為阿里云原創內容，未經允許不得轉載。?

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。