跳轉到主要內容
工程的博客

查詢從轉眼間三角洲湖表和雅典娜,改進操作並發性,和合並的性能

如來佛Das
丹尼李
通過如來佛Das和丹尼李

2020年1月29日 在工程的博客

分享這篇文章

得到一個O ' reilly的新電子書的早期預覽一步一步的指導你需要開始使用三角洲湖。

我們興奮地宣布釋放三角洲湖0.5.0,轉眼間介紹/雅典娜支持和改進並發性。

這個版本的主要特點是:

  • 支持其他處理引擎使用manifest文件(# 76)——你現在可以查詢從轉眼間三角洲表和亞馬遜雅典娜使用manifest文件,您可以生成使用Scala, Java、Python和SQL api。看到轉眼間,雅典娜三角洲湖集成文檔細節
  • 提高所有三角洲湖操作的並發性(# 9,# 72,# 228)- - -現在你可以同時運行多個三角洲湖操作。三角洲湖的樂觀並發控製已得到改進,使衝突檢測更細粒度。這使它更容易三角洲表上運行複雜的工作流。例如:
    • 並發運行刪除(例如GDPR合規)在舊分區而更新的分區被附加。
    • 並發運行更新和合並分離集的分區。
    • 並發運行的文件件和附加(見下文)。

有關更多信息,請參閱開源三角洲湖0.5.0發行說明。在這篇文章中,我們將詳細說明閱讀三角洲湖表轉眼間,改進操作並發性,更容易和更快的數據使用純插入合並重複數據刪除。

閱讀與轉眼間三角洲湖表

中描述的簡單,可靠的插入和刪除三角洲湖表上使用Python api,修改刪除等數據通過選擇性地寫新版本的文件包含的數據被刪除,隻有標誌著先前的文件刪除。這種方法的優點是,三角洲湖使我們能夠穿越時間(即。時間旅行)和查詢以前的版本。

了解哪些文件(行)包含最新數據,默認情況下您可以查詢事務日誌(更多信息深入三角洲湖:開箱事務日誌)。其他係統如轉眼間和雅典娜能讀一個生成的清單文件,文本文件包含的數據文件讀取列表查詢一個表。要做到這一點,我們將遵循Python指令;有關更多信息,請參考設置轉眼間或雅典娜三角洲湖三角洲表集成和查詢。

生成三角洲湖清單文件

讓我們首先創建三角洲湖清單文件,下麵的代碼片段。

deltaTable = DeltaTable.forPath (pathToDeltaTable)deltaTable.generate (“symlink_format_manifest”)

顧名思義,這個表中生成清單文件的根文件夾。如果您已經創建了departureDelays表每簡單,可靠的插入和刪除三角洲湖表上使用Python api表中,你將有一個新的文件夾根文件夾:

美元/ departureDelays.delta / _symlink_format_manifest

用一個文件命名清單。如果你回顧清單內的文件(如貓清單),你會得到以下輸出指示的文件包含最新的快照。

文件:$ / departureDelays.delta / - 00003……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00006……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00001……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00000……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00000……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00001……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00002……-c000.snappy.parquet一部分文件:$ / departureDelays.delta / - 00007……-c000.snappy.parquet一部分

創建轉眼間表讀取清單文件生成的

下一步是創建一個外部表在蜂房裏Metastore這樣轉眼間(或雅典娜用膠水)可以讀取生成的清單文件識別哪些鋪文件閱讀閱讀最新的三角洲表的快照。注意,轉眼間,您可以使用Apache火花或蜂巢CLI運行以下命令。k。

1。創建外部表departureDelaysExternal (…)2。行格式SERDE“org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe”3所示。存儲作為INPUTFORMAT4所示。OUTPUTFORMAT“org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat”5。位置“$ / departureDelays.delta / _symlink_format_manifest”

一些重要的筆記模式執行:

  • 模式定義在1號線必須匹配模式的三角洲湖表(例如,在這個例子中,departureDelaysExternal)。注意,分區方案是可選的。
  • 行5點的位置清單文件的形式/ _symlink_format_manifest /

SymlinkTextInputFormat配置轉眼間(雅典娜)拚花的列表數據文件的清單文件,而不是使用目錄清單。注意,對於分區表,有需要執行每額外的步驟配置很快閱讀生成的體現。

更新清單文件

重要的是要注意,每次更新的數據,您將需要重新生成清單文件所以很快將能夠看到最新的數據。

改善操作的並發

用以下的請求,你現在可以運行更多的三角洲湖並發操作。細粒衝突檢測,這些更新更容易運行複雜工作流等三角洲表:

  • 並發運行刪除(例如GDPR合規)在舊分區而更新的分區被附加。
  • 並發運行的文件件附加。
  • 並發運行更新和合並分離集的分區。

並發附加用例

例如,通常有一個ConcurrentAppendException扔在當並發事務的並發合並操作將記錄添加到相同的分區。

/ /目標“deltaTable”是分區按日期和國家deltaTable。作為(“t”).merge (源。作為(“s”),”年代。user_id = t。user_id和s。日期= t。日期和年代。國家=t.country").updateAll .whenMatched () ().insertAll .whenNotMatched () (). execute ()

上麵的代碼片段有可能導致衝突,因為條件不夠明確,即使表已經分區按日期和國家。目前的問題是,查詢將掃描整個表可能導致衝突與並發操作更新其他分區。通過指定specificDate和specificCountry所以你可以在一個特定日期或國家合並,現在這個操作是安全的並發運行在不同的日期和國家。

/ /目標“deltaTable”是分區按日期和國家deltaTable。作為(“t”).merge (源。作為(“s”),”年代。user_id = t。user_id和d。日期= '”+ specificDate +“和d。國家='"+ specificCountry +“”).updateAll .whenMatched () ().insertAll .whenNotMatched () (). execute ()

這種方法是相同的所有其他三角洲湖操作(例如刪除元數據改變,等等)。

並發文件壓縮

如果你不斷寫數據到三角洲表,隨著時間的推移將會積累大量的文件。這是特別重要的在流場景在小批量添加數據。這導致文件係統繼續積累很多小文件;隨著時間的推移這將降低查詢性能。一個重要的優化任務是定期取大量的小文件和重寫他們少數較大的文件,即文件壓縮。

在過去,有一個更高的潛在異常時並發查詢數據和運行文件壓縮。但是,由於這些改進,還可以運行查詢(包括流媒體查詢)和文件壓縮同時沒有任何異常。例如,如果您的表分區,你想重新分區隻有一個分區基於謂詞,可以隻讀分區使用,和寫回使用replaceWhere:

路徑=“…”分區=“年= 2019”numFilesPerPartition=16#緊湊分區的一個表來不。的文件
              (spark.read.format(“δ”).load(路徑)。在哪裏(分區).repartition (numFilesPerPartition).write.option (“dataChange”,“假”).format(“δ”).mode(“覆蓋”).option (“replaceWhere”,分區).save(路徑))

注意,使用dataChange = =假選項隻有當沒有數據發生變化時(比如在前麵的代碼片段)否則這可能腐敗的底層數據。

更容易和更快的數據使用純插入合並重複數據刪除

一個共同的ETL用例是收集日誌和附加到三角洲湖表。一個常見的問題是,源生成重複的日誌記錄。與三角洲湖合並,可以避免插入這些重複的記錄,如以下代碼片段涉及合並飛行數據更新。

#合並merge_table與航班deltaTable.alias \(“飛行”)。合並(merge_table.alias(“更新”)、“航班。日期= updates.date ") \.whenMatchedUpdate (集={“延遲”:“updates.delay”}) \.whenNotMatchedInsertAll () \。執行()

δ0.5.0湖之前,是不可能讀deduped數據作為流從三角洲湖表因為純插入合並沒有純粹的添加到表中。

例如,在一個流媒體查詢,您可以運行一個合並操作foreachBatch不斷編寫任何流數據到三角洲湖與重複數據刪除表中提到以下PySpark片段。

從三角洲。表s import*deltaTable=DeltaTable。forPath(火花,“/數據/聚合”)#函數來upsert microBatchOutputDF成δ表使用合並def upsertToDelta (microBatchOutputDF batchId):deltaTable.alias (“t”)。合並(microBatchOutputDF.alias (“s”),”年代。關鍵\ = t.key”).whenMatchedUpdateAll () \.whenNotMatchedInsertAll () \。執行()}#寫輸出的流媒體聚合查詢成δ表streamingAggregatesDF。writeStream \.format \(“δ”).foreachBatch upsertToDelta \.outputMode \(“更新”)。開始()

在另一個流媒體查詢,你可以不斷地從這個三角洲湖表讀取刪除處理數據。這是有可能的,因為純插入合並-介紹了δ0.5.0湖三角洲表隻會添加新的數據。

開始使用δ0.5.0湖

嚐試嚐試今天三角洲湖的前麵的代碼片段在Apache火花2.4.3(或更新)實例。通過使用三角洲湖,您可以使您的數據湖泊更可靠(不管你創建一個新的或現有的數據遷移湖)。要了解更多,請參考https://delta.io/並加入三角洲湖開源社區通過鬆弛和穀歌集團。你可以跟蹤所有的即將推出,計劃特性三角洲湖github的裏程碑。

學分

我們想要感謝下列貢獻者更新,doc變化,和貢獻在三角洲湖0.5.0:安德魯•福格蒂Andreas Neumann Burak•,丹尼李,法比奧·b·席爾瓦JassAbidi,馬修權力,成員Mukul沒吃,尼古拉斯巴黎,Pranav Anand,拉胡爾從詼諧,雷諾鑫,Shixiong朱,如來佛Das,托馬斯Bartalos,小李。

免費試著磚
開始
看到所有工程的博客的帖子
Baidu
map