一、觀遠(yuǎn)數(shù)據(jù)分析產(chǎn)品簡(jiǎn)介

觀遠(yuǎn)數(shù)據(jù)成立于2016年，總部位于杭州，主要為企業(yè)提供一站式的數(shù)據(jù)分析與智能決策產(chǎn)品和解決方案，客戶包括聯(lián)合利華、招商銀行、安踏、元?dú)馍?、小紅書(shū)、B站等，分布在零售、消費(fèi)、金融、互聯(lián)網(wǎng)等各個(gè)領(lǐng)域。公司的愿景是——“讓業(yè)務(wù)用起來(lái)，讓決策更智能”。我們發(fā)現(xiàn)很多時(shí)候數(shù)據(jù)分析產(chǎn)品并沒(méi)有很好地在企業(yè)內(nèi)部被用起來(lái)，往往是業(yè)務(wù)提需求，讓IT部門(mén)做一些數(shù)據(jù)處理和報(bào)表，這個(gè)周期可能會(huì)比較長(zhǎng)，并不利于敏捷、及時(shí)的分析決策。

關(guān)于數(shù)據(jù)分析產(chǎn)品功能，觀遠(yuǎn)已經(jīng)做的比較完備，涵蓋了數(shù)據(jù)接入、數(shù)據(jù)開(kāi)發(fā)、數(shù)據(jù)分析、數(shù)據(jù)應(yīng)用等各個(gè)環(huán)節(jié)。產(chǎn)品圍繞著“讓業(yè)務(wù)用起來(lái)”，在易用性方面有一些比較有特色的功能，比如智能ETL，它也是最受客戶歡迎的功能之一，使用門(mén)檻非常低，業(yè)務(wù)人員可以不用關(guān)心SQL，通過(guò)拖拉拽的方式進(jìn)行一些數(shù)據(jù)算子的組合來(lái)進(jìn)行數(shù)據(jù)開(kāi)發(fā)。另外我們基于Delta lake和Python開(kāi)發(fā)了數(shù)據(jù)解釋功能，提供了對(duì)數(shù)據(jù)進(jìn)行多維分析的能力，可以去探尋數(shù)據(jù)背后的根因，提升我們對(duì)數(shù)據(jù)的洞察能力。

這里舉一個(gè)具體的客戶案例，某頭部銀行，其BI平臺(tái)月活達(dá)到4萬(wàn)以上，百分之九十的分析行為可以在3-5秒內(nèi)完成。他們的計(jì)算引擎構(gòu)建在18000核的超大集群上，每日完成超過(guò)50萬(wàn)的Spark任務(wù)。能夠支撐這么多用戶活躍地使用，背后是依托于Delta Lake和Spark的存儲(chǔ)計(jì)算方案。接下來(lái)將介紹我們?cè)谶@一領(lǐng)域的一些實(shí)踐。

二、Delta Lake的應(yīng)用實(shí)踐

1、數(shù)據(jù)湖架構(gòu)介紹

Delta Lake是Databricks公司開(kāi)源的數(shù)據(jù)湖存儲(chǔ)方案，最初選擇Delta Lake的一個(gè)原因也是因?yàn)槭荄atabricks開(kāi)源，和spark的結(jié)合應(yīng)該會(huì)有比較好的性能。

這是一個(gè)比較典型的架構(gòu)圖，它依托于底層的 HDFS 或者是對(duì)象存儲(chǔ)，又或者云上的一些存儲(chǔ)方案，去支撐上層的BI、AI應(yīng)用，對(duì)多應(yīng)用的支持也是數(shù)據(jù)湖的一個(gè)重要的特性。數(shù)據(jù)入湖的方式包括批量和流式。在BI的分析場(chǎng)景，批量方式會(huì)多一些，實(shí)時(shí)能力也在逐漸興起。

接下來(lái)我們看一下將 Delta應(yīng)用在BI平臺(tái)之后的整體架構(gòu)。數(shù)據(jù)接入層負(fù)責(zé)將客戶的數(shù)倉(cāng)、業(yè)務(wù)系統(tǒng)、文件、API通過(guò)全量、增量、CDC的方式，接入到我們的平臺(tái)中。數(shù)據(jù)存儲(chǔ)和管理負(fù)責(zé)Delta Lake數(shù)據(jù)集、元數(shù)據(jù)管理，權(quán)限管理、血緣管理、數(shù)據(jù)質(zhì)量檢查以及一些優(yōu)化策略等。

數(shù)據(jù)處理和調(diào)度部分，離線開(kāi)發(fā)主要基于Spark進(jìn)行，它作為核心計(jì)算引擎，同時(shí)也支持機(jī)器學(xué)習(xí)工具，如Pandas，delta-rs 等。我們也引入了 ClickHouse作為查詢加速引擎，和Spark在某些場(chǎng)景形成互補(bǔ)。使用DolphinScheduler 作為任務(wù)編排和調(diào)度工具。數(shù)據(jù)分析和應(yīng)用層包括BI平臺(tái)、數(shù)據(jù)科學(xué)平臺(tái)以及一些數(shù)據(jù)應(yīng)用等。

2、Delta Lake的特性及應(yīng)用

Delta Lake 的重要特性包括ACID事務(wù)的支持、全量/增量更新、Schema管理、對(duì)多引擎的支持（包括spark、機(jī)器學(xué)習(xí)框架等）、數(shù)據(jù)版本支持、分區(qū)、存算分離適配多種存儲(chǔ)方案以及流批一體的能力。

首先我們來(lái)看一下Delta Lake的表結(jié)構(gòu)，其中 delta_log目錄用來(lái)記錄對(duì)表的變更歷史。每次commit都會(huì)生成一個(gè)JSON文件，每10次提交會(huì)生成一個(gè) checkpoint文件。為什么會(huì)有checkpoint文件？它可以在Spark讀取數(shù)據(jù)的時(shí)候提供一些性能優(yōu)化。當(dāng)通過(guò)Spark去訪問(wèn)時(shí)，可以基于某一個(gè)checkpoint文件以及之后的變更，不用去遍歷以往的大量的JSON 文件，從而提高訪問(wèn)效率。如果設(shè)置分區(qū)字段，我們就會(huì)看到類似于 date=2019-01-01這樣的文件夾，它表示在date字段上設(shè)置了分區(qū)，目錄下的parquet文件就是分區(qū)中的數(shù)據(jù)。如果沒(méi)有設(shè)置分區(qū)，這些parquet文件就會(huì)以平鋪的方式進(jìn)行組織。

接下來(lái)介紹一下ACID。首先，原子性方面，通過(guò)delta log來(lái)進(jìn)行控制和管理。在一個(gè)事務(wù)中，數(shù)據(jù)文件會(huì)被寫(xiě)入到數(shù)據(jù)文件夾下。當(dāng)事務(wù)完成時(shí)，會(huì)向delta log寫(xiě)入一條新的記錄，其中包括在事務(wù)中所有被修改的文件路徑，每一次提交都會(huì)增加表的版本號(hào)。操作過(guò)程中可能會(huì)發(fā)生異常。如果數(shù)據(jù)文件已經(jīng)被寫(xiě)入到文件夾下，當(dāng)事務(wù)失敗時(shí)，這些文件將不會(huì)作為表的數(shù)據(jù)文件。一致性采用樂(lè)觀并發(fā)控制的方式，會(huì)將寫(xiě)操作分為三階段。首先是讀取最新版本檢查哪些文件需要修改，接著開(kāi)始寫(xiě)入數(shù)據(jù)文件，最后是驗(yàn)證和提交。在這個(gè)階段中，會(huì)檢查所有將要提交的變更和其他并發(fā)的事務(wù)有沒(méi)有沖突，如果沒(méi)有沖突，就可以進(jìn)行提交，生成一個(gè)新的版本，寫(xiě)操作完成。Delta Lake默認(rèn)隔離級(jí)別是寫(xiě)序列化，結(jié)合上面介紹的樂(lè)觀并發(fā)控制策略，可以提供比較好的一個(gè)吞吐能力。最后，因?yàn)?Delta Lake表是存在 HDFS、S3 或者 NAS 這些存儲(chǔ)方案上，這些存儲(chǔ)服務(wù)本身也提供了高可用和持久化的能力，因此它的持久性是依托于底層的存儲(chǔ)服務(wù)來(lái)實(shí)現(xiàn)的。

當(dāng)我們對(duì)數(shù)據(jù)集進(jìn)行并發(fā)更新的時(shí)候，尤其是并發(fā)修改可能會(huì)涉及到相同的數(shù)據(jù)文件時(shí)，仍然有可能會(huì)發(fā)生并發(fā)異常。BI業(yè)務(wù)的特點(diǎn)是平臺(tái)上每天都會(huì)運(yùn)行大量的任務(wù)，當(dāng)出現(xiàn)異常的時(shí)候，不僅會(huì)影響當(dāng)前的任務(wù)，也會(huì)影響后面的任務(wù)。我們可以基于業(yè)務(wù)特點(diǎn)，通過(guò)一些優(yōu)化來(lái)避免這些影響。比如可以對(duì)每個(gè)表維護(hù)一個(gè)寫(xiě)操作的隊(duì)列，去順序執(zhí)行。這里面的操作包括更新、小文件合并、版本清理等可能會(huì)出現(xiàn)并發(fā)異常的操作。小文件合并和版本清理也是性能優(yōu)化的重要手段。

全量增量更新的能力，也是 BI業(yè)務(wù)中的基本能力。其中全量覆蓋用在表初次加載或者重建的時(shí)候。增量更新也是一個(gè)非常重要的特性，當(dāng)我們以 t +1 的方式從客戶系統(tǒng)中取數(shù)時(shí)，可以使用基于時(shí)間戳的增量更新機(jī)制來(lái)提高加載效率。Delta Lake也支持追加新數(shù)據(jù)的方式，這時(shí)候不會(huì)對(duì)歷史數(shù)據(jù)做修改。

數(shù)據(jù)入湖的方式，對(duì)于數(shù)據(jù)倉(cāng)庫(kù)或者業(yè)務(wù)庫(kù)來(lái)說(shuō)，通常可以采用JDBC 的方式從源端抽數(shù)生成一個(gè)臨時(shí)文件，通過(guò) Spark 把它轉(zhuǎn)化為Delta Lake的數(shù)據(jù)集。這種方式對(duì)接方便，較為通用。針對(duì)抽取 Hive 表比較慢的情況，因?yàn)?Hive 表本身的數(shù)據(jù)文件以及Delta Lake的數(shù)據(jù)文件都會(huì)存在 HDFS 上，沒(méi)有必要再去生成臨時(shí)文件，而是可以通過(guò) Spark 去直接加載 Hive 表并進(jìn)行轉(zhuǎn)換，這樣就大大提高了加載效率。

衍生的一個(gè)問(wèn)題是，客戶希望可以將 Hive 表和Delta Lake表進(jìn)行直接的ETL。我們的做法如下圖所示，依然通過(guò) Spark 來(lái)加載 Hive 表和 Delta Lake表，并進(jìn)行處理。前文中介紹過(guò)智能 ETL 這一模塊，我們也在不斷將這些能力和ETL進(jìn)行結(jié)合，提升數(shù)據(jù)開(kāi)發(fā)的易用性。

Schema 管理也是數(shù)據(jù)開(kāi)發(fā)中常被討論的一個(gè)話題。默認(rèn)情況下Delta Lake不允許追加 schema不匹配的數(shù)據(jù)。不過(guò)它也提供了一種機(jī)制來(lái)支持schema變化，比如通過(guò) mergeSchema參數(shù)。舉個(gè)例子，假設(shè)初始的源表有兩個(gè)字段，分別是first_name和age，通過(guò)它去創(chuàng)建一個(gè)目標(biāo)的Delta Lake表會(huì)和源表具有相同的 schema信息。當(dāng)源表的Schema發(fā)生變更，比如age字段被刪除，添加了一個(gè)新的字段，這時(shí)如果我們?cè)傧蛑暗哪莻€(gè)表去追加數(shù)據(jù)時(shí)，會(huì)發(fā)生錯(cuò)誤。我們可以使用 mergeSchema的機(jī)制來(lái)向目標(biāo)表去追加新數(shù)據(jù)，這樣目標(biāo)表schema 也會(huì)反映源表的變更。

多引擎支持方面，Spark 是我們核心的處理引擎，它是和Delta Lake結(jié)合得最緊密，也是使用場(chǎng)景最廣泛的計(jì)算引擎。我們的產(chǎn)品中使用 Spark 和Delta Lake來(lái)進(jìn)行數(shù)據(jù)的接入、開(kāi)發(fā)、數(shù)據(jù)分析等任務(wù)。Spark本身作為大數(shù)據(jù)技術(shù)中的明星，有很多優(yōu)異的特性，包括對(duì)大規(guī)模集群的支持、高效任務(wù)處理、社區(qū)活躍以及性能表現(xiàn)優(yōu)越等。Delta-rs是我們?cè)谒惴▽?shí)驗(yàn)中比較重要的一個(gè)工具，他是一個(gè)rust庫(kù)，上層也實(shí)現(xiàn)了python API。引入delta-rs的主要原因是避免啟動(dòng)一個(gè)很重的spark應(yīng)用，而是把大部分計(jì)算資源預(yù)留給E TL和交互式查詢使用，而且算法任務(wù)中希望通過(guò)python直接消費(fèi)數(shù)據(jù)，delta-rs剛好提供了這樣的能力。相比spark，他有更好的讀取性能。不過(guò)缺點(diǎn)是對(duì)寫(xiě)入支持不完善，存在bug，我們也對(duì)社區(qū)提了一些優(yōu)化。

Standalone Reader 是一個(gè)java庫(kù)，可以比較方便的讀取數(shù)據(jù)、和schema。不過(guò)這個(gè)項(xiàng)目感覺(jué)缺乏維護(hù)，存在一些bug。也不支持sql查詢和寫(xiě)入。使用場(chǎng)景比較有限，比如說(shuō)數(shù)據(jù)集的簡(jiǎn)單預(yù)覽等。我們也對(duì)這個(gè)項(xiàng)目做了一些修復(fù)。

時(shí)間旅行，也叫 time travel，指的是對(duì)數(shù)據(jù)多版本的支持。給大家介紹一個(gè)算法實(shí)驗(yàn)中的場(chǎng)景，比如我們的算法工程師在進(jìn)行算法實(shí)驗(yàn)時(shí)，除了使用最新的數(shù)據(jù)外，也想使用歷史版本的數(shù)據(jù)來(lái)進(jìn)行實(shí)驗(yàn)效果的比對(duì)，時(shí)間旅行剛好可以很好地解決這些問(wèn)題。

關(guān)于分區(qū)，也以一個(gè)算法實(shí)驗(yàn)的場(chǎng)景來(lái)說(shuō)明如何通過(guò)分區(qū)來(lái)提高算法實(shí)驗(yàn)的效率。我們和客戶合作的銷量預(yù)測(cè)場(chǎng)景，業(yè)務(wù)上區(qū)分了不同的產(chǎn)品線，比如洗浴產(chǎn)品、食品等。每條產(chǎn)品線的業(yè)務(wù)形態(tài)不同，因此需要去分別構(gòu)建模型。不過(guò)這些特征工程的邏輯比較接近，所以我們可以把這些都放在一個(gè)數(shù)據(jù)集里，通過(guò)分區(qū)來(lái)管理。因?yàn)椴煌謪^(qū)的寫(xiě)入不會(huì)發(fā)生沖突，這些流程都可以并發(fā)運(yùn)行，從而提升了的執(zhí)行效率。對(duì)分區(qū)字段的選擇，也是有一些要求的，比如一般是date類型，在進(jìn)行正式的分區(qū)之前，可以做一些前置檢查看字段是否適合分區(qū)，避免使用錯(cuò)誤的情況；另外，分區(qū)適合對(duì)大表使用。

下圖是一個(gè)典型的流式寫(xiě)入流程，通過(guò)一些實(shí)時(shí)同步工具，將源端數(shù)據(jù)同步到Kafka，再通過(guò)Spark Structured Streaming進(jìn)行增量更新，同步到Delta Lake，供上層應(yīng)用使用。

性能優(yōu)化主要在以下四個(gè)方面：

• 小文件合并（compaction）：當(dāng)持續(xù)更新數(shù)據(jù)集時(shí)，數(shù)據(jù)文件會(huì)不斷增多，我們觀察發(fā)現(xiàn)，當(dāng)對(duì)一個(gè)數(shù)據(jù)集進(jìn)行比較高頻的更新，比如每 5 分鐘做一次更新，在幾個(gè)小時(shí)之內(nèi)，文件數(shù)量就可能增加到數(shù)萬(wàn)甚至更多。大量的文件會(huì)嚴(yán)重影響 Spark 的查詢性能。因此我們需要將大量的小文件壓縮為少量大文件，去提高訪問(wèn)效率。
• 版本清理（vacuum）：在數(shù)據(jù)分析的很多場(chǎng)景我們只需要使用最新數(shù)據(jù)就可以了，Delta Lake提供了vacuum機(jī)制來(lái)進(jìn)行版本清理。如果使用了時(shí)間旅行，需要根據(jù)情況來(lái)制定版本清理策略。關(guān)于小文件合并和歷史版本清理的時(shí)機(jī)，我們一般通過(guò)定時(shí)任務(wù)來(lái)觸發(fā)，我們也在客戶環(huán)境觀察到這樣的現(xiàn)象，比如一天之內(nèi)會(huì)有幾個(gè)任務(wù)比較密集的時(shí)間段，文件和磁盤(pán)的增長(zhǎng)都比較快，我們的定時(shí)策略可以靈活地調(diào)整，在這些任務(wù)完成之后進(jìn)行及時(shí)地合并和清理。
• 僅讀取需要的列：算法中也有一些典型應(yīng)用場(chǎng)景，比如可以把Delta Lake的表當(dāng)成特征庫(kù)來(lái)使用，構(gòu)建包含數(shù)百列的大寬表，但每個(gè)產(chǎn)品線在建模時(shí)，可以只選擇其中所需要的列，這也利用了列式存儲(chǔ)的優(yōu)勢(shì)。
• 持續(xù)升級(jí)版本

三、總結(jié)和展望

我們會(huì)持續(xù)地去升級(jí)版本，使用新的一些特性，比如Z-Order對(duì)查詢性能的優(yōu)化，DML增強(qiáng)等。同時(shí)，我們也會(huì)讓我們的產(chǎn)品更加云原生，融合多引擎，包括 Databricks，ClickHouse等。也計(jì)劃讓Delta Lake更加開(kāi)放，可以通過(guò) SQL 的方式提供給其它工具使用。此外，我們計(jì)劃可以基于數(shù)據(jù)集catalog做數(shù)據(jù)資產(chǎn)管理。我們也在持續(xù)的回饋社區(qū)，公司內(nèi)有多位delta lake，spark，delta-rs的contributor。