阿里云開發者社區ALIBABACLOUDDEVELOPERCOMMU從入門到實踐從技術基礎到實踐應用首月599元試用阿里云開發者“藏經閣”湖Lakehouse以及部分spark相關應用。書的應用環境/product/bigdata/spark基礎篇一、Databricks數據洞察一企業級全托管Spark大數據分析平臺 1二、DeltaLake的演進歷程和現狀優勢 三、深度解析數據湖存儲方案Lakehouse架構 四、DeltaLake數據湖基礎介紹(開源版) 五、DeltaLake數據湖基礎介紹(商業版) 應用篇 55七、使用DDI+Confluent進行實時數據采集入湖和分析 八、使用Databricks進行零售業需求預測的應用實踐 九、使用Databricks進行營銷效果歸因分析的應用實踐 十、使用Databricks和MLflow進行機器學習模型訓練和部署的應用實踐 87分析平臺作者：棕澤Databricks數據洞察產品介紹1)ApacheSpark創始公司，也是Spark的最大代碼貢獻者，Spark技術生態背后的商業公司。在2013年，由加州大學伯克利分校AMPLab的創始團隊ApacheSpark的創建者所成立。ApacheSpark、DeltaLake、Koalas、MLFlow、OneLakehousePlat3)公司定位⑩DatabricksistheData+Alcompany,為客戶提供數據分析、數據工程、數據科學和一體化分析平臺4)市場地位⑩科技獨角獸，行業標桿，領導Spark整體技術生態的走向及風向標⑩2021年最受期待的科技上市公司(來源Databricks官網)企業級全托管Spark大數據分析平臺及案例分析21)2019年10月G輪，估值$6.2Billion2)2021年2月初F輪，估值$28Billion350億美元，甚至是高達500億美元Databricks從成立之初的融資/估值歷史F輪62億美元Sep20133.Databricks和阿里云聯手打造的高品質Spark大數據分析平臺企業級全托管Spark大數據分析平臺及案例分析3NICHEPLAYERSDataiku●4.Databricks+阿里云=Databricks數據洞察1)產品核心2)產品引擎與服務⑩100%兼容開源Spark,經阿里云與Databricks聯合研發性能優化企業級全托管Spark大數據分析平臺及案例分析4DeltaEngine(超高性能)容ApacheSpark)pythonspok4)產品關鍵信息與優勢免運維：無需關注底層資源情況，降低運維成本，聚高品質支持：阿里云和Databricks提供覆蓋Spark高性能分析：性能提升3-50倍提升，輕松處理十億級無縫架構升級：無縫替代Lambda架構，輕松實現批先進的存儲技術：支持CRUD操作、TimeTravel、學工作環境。先進的ML工作流程(待發布):基于MLfow和WorkspacDatabricks驅動的超高性能阿里云數據平臺SuperChargedAliCloudDataPlatformpoweredbyDatabricks◎全托管◎全托管高性能架構先進企業級Spark交互協作流式數據結構化數據半結構化數據Aspak·100%兼容ApacheSpark02TPC-DS10TB數據，Runtime最快50倍提升0企業級全托管Spark大數據分析平臺及案例分析6opómumublizationComputeoplimizaliondurmngoAutoscalingcomputeHighavailabityforcdusteSchemaevolution&enforcSreamingDataSink/S兩者詳細比較：htps:/helpalbyuncomdocamentdeta175748存算一體=》計算存儲分離存算一體=》計算存儲分離三NetworkNetworkNetwork按月標價(元)(元/GB/月)hps://priceproductHecs/d(元/GB/月)計算存儲分離=》計算存儲分離=》JindoFS實現架構方案力耄⑩不同作業類型工作流混合調度企業級全托管Spark大數據分析平臺及案例分析8三記錄8.豐富的數據源支持支持常見數據文件格式支持常見數據文件格式支持各種數據源(內置Connector)9.元數據管理全托管高性能引擎存算分離場景四：場景四：交互式ML訓練Databricks幫助客戶快速構建先進的LakeHouse平臺，幫助客戶降低運維和調優成本、構建批流一體的處理場景一：數倉遷移或向數據湖演進 數據校驗難)場景二：構建LakeHouse方案場景三：能力，聚集挖掘數據價值·Recommendation·Risk,Fraud,&Intrusion·CustomerA·loT&Predictive低質量、不可靠的數據差強人意的性能在這樣的背景下應運而生了DeltaLake場景4.大數據發展進入LakeHouse時代▲驅動力1.數據模型變化難2.不支持大規模非結構化數據驅動力1.不支持事物2.不支持數據質量校驗Structured,Semi-Structured,andUnstru傘3.不支持計算存分類，存儲成本高3.不支持批流一體5.使用DDI構建批流一體數倉，簡化復雜構架8KafkaStremKafkaooKatfkaKatfka企業級全托管Spark大數據分析平臺及案例分析11冒乖④世婪日□回7.Databricks數據洞察典型構架DDI與阿里云產品深度集成(典型場景)1)數據獲取3)BI報表數據分析&交互式分析4)AI數據探索⑩支持機器學習，Mllib等Spark生態Al場景。5)上下游網絡打通⑩如上游對接Kafka、OSS、EMRHDFS等等，下游承接Elasticsearch、RDS、OSS存儲等。典型場景客戶案例介紹⑩數據收集/存儲：接收實時產生的流式數據和外部云存儲上批量數據⑩數據ETL:持續高效地處理增量數據，支持數據的回滾和刪改，提供ACID事務性保障⑩BI數據分析&交互式分析：支持查詢，Notebook可視化分析，無縫對接多種BI分析工具⑩數據科學：支持機器學習/深度學習OSS存儲等◎jupyterDataScience↑Structured,Semi-structured&UnstructuredData↑Structured,Semi-structured&UnstructuredData數倉的架構，再到最近兩年的Lakehouse架構。大數據平臺架構演進-數據倉庫 同時，基于它自身的諸多優化特性，數倉架構對分析型場景能夠提供非常好的支持。但是支持的場景比較有限，基本局限于常用的分析場景。而在大數據時代，隨著數據規模的逐漸增加，企業對于數據分析的場景要求越來越多，逐漸產生了一些高級的分析場景需求，比如數據科學類或者機器學習類的場景，而數據倉庫對此類需求難以支持。另外，數據倉庫也無法支持半結構化以及非結構化的數據。阿里云大數據平臺架構演進-數據湖+數據倉庫阿里云基于Parquet等開放數據格式借助低成本的云對象存儲，支持結構和非結構類型數據存儲；SchemaSchema-On-Read設計，下游數據治理復雜度高；缺少數據庫管理特性及性能優化手段32003年前后，Hadoop面市。伴隨著數據規模體量的爆炸式增長，我們對低成本存儲的需求也愈發迫切。于是第二代大數據平臺架構雛形初現。它以數據湖為基礎，能夠支持對結構化、非結構化以及半結構化數據的存儲。與數據倉庫相比，它是一種Schema-on-read的設計，數據能夠比較高效地存入數據湖，但是會給下游的分析提供較高的負擔。因為數據在寫入之前沒有做校驗，隨著時間的推移，數據湖里的數據會變得越來越臟亂，數據治理的復雜度非常高。同時因為數據湖底層是以開放的數據格式存儲在云對象存儲上，云對象存儲的一些特性會導致數據湖架構缺少像數倉一樣的數據管理特性。另外因為云對象存儲在大數據查詢場景上的性能上不足，導致很多場景下都無法很好地體現數據湖的優勢。DeltaLake的演進歷程和現狀優勢15大數據平臺架構演進-大數據平臺架構演進-Lakehouse基于云對象存儲的數據湖之上提供ACID事務管理層，同時保證了數據質量和數據訪問性能；支持大數據時代各種分析處理場景AmaznS于是第三代大數據平臺架構——Lakehouse應運而生。它在數據湖之上抽象出了事務管理層，能夠提供傳統數倉的一些數據管理特性，還可以針對云對象存儲中的數據做一些數據的性能優化。從而能夠針對大數據時代各種復雜的分析場景提供支持，且對于流批兩種場景也能夠提供統一的處理方式。有了Lakehouse架構的背景之后，DeltaLake也應運而生，它是由Databricks開源的一個數據庫解決方案，架構清晰簡潔，能夠提供比較可靠的保障。上圖呈現了Deltalake的Multi-hopMedallion架構，即通過多個表結構來提供不同分析場景的支持。數據可以通過streaming的方式流入DeltaLake,也可以使用batch方式DeltaLake里的表可以分為三類：⑩第一類：數據最先導入的表稱為bronzetable。direct事務的特table中數據的可靠性，因此它是整個數據湖的sourceoftruth(事實表)。某些場景可以直接讀取bronzetable。DeltaLake的演進歷程和現狀優勢16阿里云阿里云底層基于事務日志機制實現可串行化的隔離級別，提供ACID事務，保證讀寫數據的一致性；在ACID事務基礎上，提供了更多數據管理及性能優化特性時間回溯時間回溯支持制約束及自動演化緩存及索引在大數據的場景下，元數據管理本身可能會成為一種負擔，因為對于較大的表來說，元數據本身就能成為大數據。所以如何高效地支持元數據管理，也是對架構挑戰。此外，在Deltalake的商業版本里，還提供了數據庫中的數據布局自動優化的能力，同時實現了傳統數倉數據庫一系列性能優化特性，比如緩存、索引等優化能力。0.1版本0.2~0.4版本0.5版本0.6版本0.7版本0.8版本1.0版本+Deltalake項目最早開源在2019年4月，事務、流批一體等最核心的功能在0.1版本都社區第一次在Spark之外提供引擎的支持，也是Deltalake開放性目標的一部分；同時0,5版本還提供了一些優化的特性，以及通過SQL的方式直接將parque轉成⑩0.6版本：DeltaLake做了一些了進一步優化，對比如describehistory的命令提供了更多metrics信息。元數據本身是transactionog事務日志的一部分，所以有了Hivemetastore的支了對dml的支持。支持了VACUUM的并發刪除能力。縱觀Deltalake的發展歷程，可以清晰地看出，它一直堅定地朝著Everywhere——支持更多元、更開放的生態發展。Support上圖展示了Deltalake1.0的一些核心特性。阿里云阿里云1……本身是時間戳，但是數據中并沒有eventDate⑩方案1:直接使用時間戳的字段做分區。時間戳是一個比較細粒度的字段，使用它來做會產生大量的分區，對于查詢性能會造成非常大的影響。因此，此方案被排除。⑩方案2:數據寫入表之前手動維護額外字段。比如從eventTime字段中抽取得到eventDate。但這需要人工維護字段，而但凡涉及到人工，就容易引入錯誤，尤其是多種數據源同時寫入的情況，要求對多種數據源同時做轉換，極易出現差錯。因此DeltaLake1.0提供了generatedcolumns,它是一種特殊類型的列，它的值可以根據用戶指定的函數自動生成。阿里云阿里云eventDateGENERATEDALWAYSAS)eventDateGENERATEDALWAYSAS)1…PARTITIONEDBY(eventDate)可以通過如上圖簡單的SQL語法，將eventTime轉換成date類型，從而生成eventDate字段。整個過程自動完成的，用戶只需要在最開始創建表的時候提供這個語法即可。Deltalake1.0提供的第二個重要特性是Standalone。它的目標是可以在Spark之外對接更多引擎，但是諸如Presto、Flink等引擎本身并不需要依賴Spark,如果Deltalake只能強綁定Spark就違背了Deltalake開放性的目標。DeltaLake的演進歷程和現狀優勢20√√√√√√√7√√√√0InIn[2]:fromdeltalake['part-80000-087ecfo?-ee64-49b-bebb-f6c5d6c944F8-c200,snapp'part-0000-02ea?B3d-c9b5-4461-a1e0-526595f1a10-c?0.snappy.pa'part-00090-049b?14d-10a8-4aab-bf3b-cd?aelace?91-cee?.snappy.p'part-0098-06a6346b-c533-4cda-83a8-bc514b749761-cee0.snappy.pa最早的DeltaLake是Spark的一個子項目，因此DeltaLake對Spark引擎的兼容性做得非常好。同時，由于Spark社區發展迅速，能夠第一時間兼容Spark也是DeltaLake社區的首要目標。所以在1.0版本，DeltaLake首先兼容了Spark3.1,并對其提供的一些特性進行優化，以便第一時間在DeltaLake里投入使用。很多企業在使用DeltaLake的時候，一個常用場景是使用單一集群去訪問/關聯一個存儲系統。有DeltaLake1.0對此提供了delegatinglogstore功能，通過logstore的方式來支持不同云廠商的對象存儲系統，以便能夠支持混合云部署的場景，同時也可以避免對單一云服務商產生locking綁定的情況。未來，DeltaLake社區會朝著一個更加開放的方向發展。染基于數據布局的優化-OPTIMIZE&ZOREDERBY這兩個特性在Databricks目前開Spark產品-Databricks數據洞察，除了此處列出的Optimize及Z-Ordering,可以在DeltaLake的演進歷程和現狀優勢23阿里云Databricks數據洞察產品架構⑨⑨全托管Spark平臺spankDatabricks數據洞察產品是基Databricks的引擎提供的全托管數據平臺，它最核心的部引擎。相比于開源的Spark和Delta,商業版在性能上有非常大的提升。DeltaDeltaLake當前應用情況最后，看一下DeltaLake當前在全球范圍內的應用情況，越來越多的企業已經開始使用DeltaLake來構建Lakehouse。從Databricks給的數據看，目前已有超過3000+客戶在生產環境中部署了DeltaLake,每天處理Exabytes級別數據量，其中超過75%的數據已流式分析BI結構化，半結構化及非結構化數據?數據科學機器學習數據倉庫數據倉庫優勢·便于BI和報表系統接入·數據管控能力強劣勢·不支持非結構化數據·成本高，靈活度低潮目片、文檔等的支持是非常有限的，因此它不適用于類似于機器學習的應用場景。而且一般情況下，數據倉庫都是專有系統，使用成本比較高，數據遷移和同步的靈活性比較低。因此，為了解決上述問題，數據湖的架構應運而生。8綜上，不論是數據倉庫還是數據湖，都無法完全滿足用戶的需求。復雜的組合型數據系統數據倉庫數據工程流式處理HadooAgecheAunArushekafaApacheSpApsheFinkArazenKiaAneSremAndytiesGoger3十3十)8警8因此，在很多實際使用場景中，用戶會將兩者組合起來使用，但是這導致需要構建很多不同的技術棧來支持所有場景。比如對于數據分析，需要將結構化的數據輸入到數據倉庫，然后建立數據市場，對數據分析和BI提供支持；對于數據科學和機器學習的場景，需要把不同的數據比如結構化、半結構化以及非結構化的數據存儲到數據湖中，經過數據清理，用來支持機器學習和數據科學等場景；而對于流式數據源，需要通過流式數據引擎存儲到實時數據庫中，同時還需要對數據湖里的數據進行ETL提取、轉換和加載，來保證數據的質量。而且，由于所有技術棧都是互相獨立的，導致了維護和使用這些系統的團隊也是分散的。比如，數據分析師主要使用數據倉庫系統，而數據科學家主要使用數據湖系統，同時數據工程師也需要維護整個系統的不同團隊，溝通成本比較高。此外，系統中維護了很多不同格式的數據副本，沒有統一的管理數據模型，不同團隊的數據很有可能會產生差異。①①數據科學機器學習6:流式分析BI解決的辦法是綜合數據湖與數據倉庫的能力——基于數據湖使用的對象存儲，構建數據倉庫擁有的數據管控能力。而這一切的關鍵就是其中的結構化事務層。流式分析BI此前，數據湖主要存在以下幾個痛點：務性地被同時讀到或者同時沒有讀到，而這是難以實現的，因為在分布式的對象存儲上寫多個文件，設置一個文件，數據的一致性都是不能完全被保證的。2)數據的修改。由于安全合規等原因，用戶會有強制性地修改已有數據的需求，特別是有時候需要根據過濾結果細粒度地修改某些數據。由于數據湖在數據管控能力上的不足，在數據湖上實現此需求往往需要使用全部掃描再重寫的方式，成本比較高，速度也比較慢。3)如果一個作業中途失敗，而它產生的部分數據已經存入到數據庫中，這也會導致數據的損壞。4)批流混合輸入。由于數據在批和流系統中都存在，可能會造成數據在兩套系統中不一致，導致讀取結果不一致。5)存數據歷史。有些用戶需要保證數據查詢的可重復性，方案之一是為了這個需求做很多重復的數據快照，但這會導致數據的存儲和計算成本都大幅上升。6)處理海量的元數據。大型數據湖元數據的數據量非常大，經常能夠達到大數據的級別。很多數據湖采用的數據目錄系統無法支持如此大量的元數據，這也限制了數據湖的擴展性。7)大量小文件的問題。在數據不斷輸入的過程中，數據湖內會產生大量小文件，隨著時間的推移，小文件的數量可能會越來越多，這會嚴重影響數據湖的讀取性能。8)性能問題。在數據湖上達到高性能不是一件容易的事。有的時候為了達到一定的性能要求，用戶需要手動做一些性能的優化，比如數據分區等，而這些手動的操作又比較容易出錯。9)數據的查詢管控。由于數據湖的開放性，確保查詢權限合規也是需要解決的問題。10)質量問題。前面很多點都會導致數據質量的問題。在大數據場景下，如何確保數據的正確性也是一個普遍的問題。而DeltaLake能夠為Lakehouse帶來數據質量、可靠性以及查詢性能的提升。ACID事務事物化所有操作·ACID事務事物化所有操作·每一個操作，要么整體成功，要么整體失敗g5.保存數據歷史上述前五個問題都是關于數據可靠性，它們都可以通過DeltaLake的acid事務能力來解決。在DeltaLake上，每一個操作都是事務的，即每一個操作都是一個整體，要么整體成功，要么整體失敗。如果一個操作在中途失敗，DeltaLake會負責將其寫入的不完整數據清理干凈。具體的實現方式是DeltaLake維護了包含所有操作的一個事務日志，能夠保證數據與事務日志的一致性。如上圖，某次寫操作在某個表中添加了很多數據，這些數據被轉換成了parquet格式的兩個文件file1和file2。有了事務日志，讀操作的時候就能夠保證要么讀不到這條日志，要么同時讀到這兩條記錄，這樣就保證了讀取的一致性，解決了讀寫并行的問題。事物化所有操作·每一個操作，要么整體成功，要么整體g5.保存數據歷史此外，有了事務日志后也可以對已有數據做細粒度的修改。比如下一次寫操作對表中的某些數據進行修改，在事務日志中就會出現刪除原有文件file1和添加修改后文件file3這樣兩條記錄。同樣，在讀取的時候，這兩條記錄也會被同時讀到或者忽略，使讀取的一致性得到保證。針對第三點中途失敗的作業，DeltaLake寫入的事務性能夠保證不完整的數據不會被成功寫入。對于批流混合的輸入數據，由于Spark天然支持批流一體，在寫入時可以將批和流的數據寫入到同一張表，避免了數據冗余及不一致性。由于事務日志保存了所有操作的歷史記錄，我們可以對之前某個時間點的歷史數據進行查詢。具體實現方法是：DeltaLake可以查到歷史某個時間點對應的事務日志，并且根據歷史的事務日志進行數據重放，得到該時間點的數據狀態。這個能力被稱為“時間旅行”。那么,DeltaLake是怎樣處理海量元數據的呢?答案很簡單，使用Spark來處理。所有DeltaLake的元數據均以開源parquet的格式存儲，數據與元數據總是相伴相生，無需進行同步。使用Spark處理元數據，使得DeltaLake的元數據可以在理論上進行無限的擴展。索引機制數據的自動優化·索引機制數據的自動優化·跳過數據掃描：分區，布隆過濾器等·Z-ordering:優化多個列的存儲布局DeltaLake還采用索引的機制來優化性能，它采用分區和不同過濾器等的機制，可以跳過數據的掃描。還采用了Z-ordering的機制，可以在對某個列進行優化的同時，使其他列性能犧牲最小化。為了解決大量小文件的問題，DeltaLake還可以在后臺定期對數據布局進行自動優化。如果存儲的小文件過多，會自動的將他們合并成大文件，這解決了數據湖中小文件越來越多的問題。表級別的權限控制表級別的權限控制·提供權限設置的API·根據用戶權限，動態地對視圖進行脫敏8.性能問題對于數據查詢的管控，DeltaLake實現了表級別的權限控制，也提供了權限設置API,可以根據用戶的權限動態對視圖進行脫敏。SchemaSchema驗證和演化·所有表中的數據必須嚴格符合schema●可以在數據寫人時進行schema演化最后，DeltaLake實現了schema的驗證功能來保證數據質量。存在DeltaLake表中的所有數據都必須嚴格符合其對應的schema,它還支持在數據寫入時做schema的合并演化。當輸入數據的schema發生變化的時候，DeltaLake可以自動對表的schema進行相應的演化。流式分析BI數據科學機器學習原始輸入與數據歷史記錄業務層聚合原始輸入與數據歷史記錄業務層聚合清理、過濾、擴展后的數據BronzeSilverBronzeSilver新，還可以減少數據冗余，從而優化存儲和計算的開銷。3)開放的數據格式可以讓數據在不同系統之間的遷移更加順暢。5)批流一體。與lambda架構不同，Lakehouse能夠做到真正的批流一體，從而簡化SSSpo☆@n⑨MachineLearningDatabricks獨家優化了databricksruntime引擎，也可以理解為ApacheSpark的加強好的結合，提供了全托管高性能的企業級Spark平臺，能夠同時支持企業的商業洞察分析以及機器學習訓練等。DeltaLake數據湖基礎介紹(開源版)34四、DeltaLake數據湖基礎介紹(開源版)作者：王曉龍(筱龍)1.大數據平臺架構演進①1)第一代：數倉架構2)第二代：數據湖+數倉架構3)第三代：Lakehouse架構2.DeltaLake-運行在數據湖之上的可靠存儲層DeltaLake作為可靠的數據存儲中間層，為構建Lakehouse提供了核心支撐。適用于所有場景適用于所有場景ACID存儲層提供高可靠性及性能存儲所有類型數據的數據湖DeltaLake的核心特性是對ACID事務支持，并且基于事務日志機制，實現可串行化的隔離級別，提供ACID事務，保證讀寫數據的一致性。2)DeltaLakeStarter-DML準備通過PySparkDataframAPI創建一張Delta表，表的名稱是random_num,表中只包>>>spark.sql(>>>spark.sql("CREATET\"}\"".format('/tmp/delta_course/delta_tab接下來往表中執行幾條簡單的修改操作語句：>>>spark.sql>>>spark.sql("insertin>>>spark.sql("deletefrom3)DeltaLake文件系統目錄結構DeltaLake表的物理存儲目錄下，既包括自身的表數據文件，也包括記錄表Schema及表變化的DeltaLogs:_delta_log_delta_log000000000000000010.checkpoin_last_checkpointpart-00000-05518637-0919-4a45-a95c-8066fa910681-c000.spart-00000-1eef41d1-febd-40fe-849c-d9b19bee3761-c000.spart-00000-258b6649-3b08-4e97-9fa?-871be15edc1a-c000.spart-00000-467f6d26-902c-4180-937c-29e?6a3e7b4e-c000.snpart-0000-60add56e-1078-40c9-a80b-5d86dffe50d5-c?00.snpart-00000-687cc4bf-4442-476b-a9a4-fec8ca89429f-c000.snpart-0000-6b042c36-59e6-44Zc-95bd-c15e45529353-c?00.snappy.parqpart-00000-85f4a3c9-12e5-4e61-b48a-bfdd6aac7496-c000.snpart-00000-9dd1e?91-fd67-4dec-887d-1b566bdcd9af-c000.spart-00000-b8c55888-fd23-46d1-bdbf-fbf0ec014c70-c000.spart-00000-c8dd38c8-7da3-472f-8171-707acab189eb-c000.snpart-00002-61073ab1-be89-4357-a3d9-d1235d475c8c-c000.spart-00004-d2f4a646-a97b-434a-bab5-6a375e41476a-c000.spart-00007-2339a4cd-24ae-4e94-a2a9-1773e29dbe61-c000.spart-00009-eb0782ff-81d2-4a44-9700-af6b8582f6fe-c000.part-00011-e708e4c4-9e2f-43a6-9950-1c9baf04c657-c000.sTransactionLog(事務日志，也稱DeltaLog)是一種有序記錄集，順序表從初始創建以來的所有事務操作。__delta_log00000000000000000010.checkp_last_checkpoint1)TransactionLog的整體設計目標，是實現單一信息源(SingleSourceofTruth),通過跟蹤記錄用戶所有的表操作，從而為用戶提供在任意時刻準確的數據視圖。2)同時，因為DeltaLake是基于ApacheSpark構建的，依托Spark來管理維護事務日志，所以相比通過Metastore使用單一的數據庫管理元數據，DeltaLake具備高可擴展的元數據處理能力，能夠支持上百PB的Delta表讀寫訪問。3)除此之外，DeltaLake的事務日志也是其它重要數據管理特性實現的基礎，比如數據版本回溯(TimeTravel)等。在DeltaLake中，Transaction被稱為Commit。每個Commit代表一個事務操作，同時也代表了一個數據版本，對應_delta_log目錄下的一個json文件。_delta_log_delta_log00000000000000000010.checkpo_last_checkpoint{"remove":{"path":"part-00004-d2f4a646-a97b-434a-bab5-6a3c000.snappy.parquet","deletionTimestamp":1645945987771,"dataChange":true,"extendedFrue,"partitionValue√事務日志的最后一行是關于commit的詳細信息，包括了時間戳、操作名等元數據。DeltaLake當前定義的Action動作包括：涉及數據文件增加和刪除(Addfile/Remove事務及協議相關的變更操作(Settransaction、·Updatemetadata-Updatesthetable'smetad·Settransactifromwhereandatwhatti⑩通過Spark獲取到表的最新狀態DeltaLake定義的Commit維護的是變更操作的過程記錄，當針對Delta表執行查詢語句時，可以通過Spark獲取到表的最新狀態。Spark會對事務日志做聚合，檢查事務日志經歷了哪些事務操作，并基于事務日志構建出可靠準確的Delta表狀態。0000000000000001json在變更操作較多的場景，比如CDC,deltalog下會生成大量json小文件，對處理性能會造成較大影響。為了解決上文提到的小文件問題，DeltaLog引入Checkpoint機制。開始到當前時刻所有變更記錄(默認每10次Commit創建一個Checkpoint文件)。①Checkpoint文件給Spark提供了一種捷徑來重構表狀態，避免低效地處理可能上千條的json格式的小文件。__delta_log0000000000000000010.checkpoint.parquet"/tmp/delta_course/delta_table/_delta_log/00000000000000000010.checkpoi-9d0b-4e...InullIDeltaLake數據湖基礎介紹(開源版)42借助checkpoint,Spark可以快只需要基于版本10及隨后的11和12兩次commit構建表的狀態，從而大大提升了元數3)樂觀并發控制DeltaLake數據湖基礎介紹(開源版)43√傳統數據庫的鎖機制其實都是基于悲觀并發控制的觀點進行實現的；√對比悲觀并發控制，樂觀并發控制可以提供更好的性能；√示例中，用戶A和用戶B都拿到版本號為0的commit,排他鎖(mutualexclusion)決定了只能有一個用戶能夠創建版本號為1的commit,假設接受了UserA的commit,操作語義，在版本1基礎上完成UserB的寫入。ACCONSISTENCYID如上文介紹，TransactionLog將事務抽象成一個個Commit(文件),Commits里可以包含不同類型的Action,但是每個Commit是原子的。MartinKleppman在DDIA書中對原子性的定義：t隔離性是針對并發事務的處理方式，并發事務不應該相互干擾。在DeltaLake中，隔離性是通過OCC+排他鎖方式實現的，并且實現了讀寫的串行化。MartinKleppman在DDIA書中對隔離性的定義：TransactionLog寫入分布式磁盤中，在事務處理結束后，對數據的修改就是永久的，即便系統故障也不會丟失。MartinKleppman在DDIA書中對持久性的定義4)一致性Consistency是數據庫的屬性，應用程序可能依賴數據庫的原子性和隔離屬性來實現一致性，但這并不取決于數據庫本身，但一致性是由應用來決定的。ACIDsense)isapropertyoftheapplication.Theapplicationmayrelyonthedatabase'satomicityandisoThus,theletterCdoesn'treally擴展的元數據處理等特性。DeltaLake的特性，敬請關注。DeltaLake數據湖基礎介紹(商業版)46五、DeltaLake數據湖基礎介紹(商業版)作者：李潔杏數據管理系統從早期的數據倉庫(DataWarehouse),已經發展到今天的Lakehouse?？茖W和機器分析的場景。個Structured,Semi-Str數據倉庫架構可以完全控制數據的存儲和查詢，因此可以同時設計查詢系統，以及適應查詢系統的數據存儲結構，以達到優越的查詢性能；而在Lakehouse架構下，數據是用開放存儲結構存儲的，如Parquet格式，以便更多系統可以便捷的訪問數據，但是開放的存儲格式并不一定適合查詢操作，查詢系統也沒有足夠的統計數據來實現高效查詢。那么,Lakehouse如何以開放的存儲格式達到高效的查詢性能?DeltaLake數據湖基礎介紹(商業版)47如圖所示，在Databricks中用SSD作為緩存，可以將數據讀取速度提高3倍以上；采用Delta引擎作為緩存，則可以將數據讀取速度提高7倍以上。2.建立輔助數據結構示例一：Parquet文件中的DataSkip查詢的信息只會存在于file3,因此可以跳過file1和file2的讀取。WHEREyear=2020示例二：在Parquet文件上建立索引如下圖，如果查詢條件是type=“DELETE_ACCOUT”,可以利用在type上建立的索引直接跳到對應的數據上，從而避免讀取無關數據。=“DELETE_ACCOUNT”示例三：Parquet文件上建立BloomFilter可以為每一個文件建立BloomFilter,BloomFilter可以快速判斷表文件中是否包含需要查詢的數據，如果不包含則快速跳過該文件，從而減少掃描數據量，提升查詢性能。=“DELETE_ACCOUNT”1)小文件問題在DeltaLake中頻繁執行MERGE,UPDATE,INSERT操作，可能會產生大量的小文件。大量的小文件，一方面會降低系統讀取性能，同時也會提高元數據操作的開銷。Lakehouse中使用了不同的技術來減少小文件的產生：個表文件進行寫入，最終會導致一個partition中產生很Databricks對Delta表的寫入過程進行了優化，對每個partition使用一個專門的executor來合并其它executor對該partition的寫入，從而避免了小文件的產生。在每次向Delta表中寫入數據之后，會檢查Delta表中的表文件數量，如果Delta表中的小文件(size<128MB則視為小文件)數量達到閾值，則會執行一次小文件合并，將Delta表中的小文件合并為一個新的大文件。除了自動合并，Databricks還提供Opitmize命令，使用戶可以手動合并小文件，優化表結構，使得表文件的結構更加緊湊。2)查詢時間問題查詢運行時間主要取決于訪問的數據量，即使使用Parquet格式，也可以通過優化表內的數據布局以減少運行時間。⑩表文件數據排序DeltaLake數據湖基礎介紹(商業版)50將表文件存儲數據排序，在每個表文件中存儲一定量的數據，如下圖中file1存儲uid=0.….1000,file2存儲uid=1001..2000,這樣在查詢時就可以根據需要跳過無關的表文件，減少文件掃描數量。 在實際查詢中，有些查詢需要看colomn1在某個范圍內的數據，有些查詢需要看colomn2在某個范圍內的數據，或者更多，這時候僅僅對colomn1進行排序顯然是不夠的。Z-Ordering可以在多個維度上(如下圖的col1-4)將關聯的信息存儲到同一組文件中，來減少不必要的文件讀取。ZZ絡4WHEREi_i查詢說明：將store_sales與item兩個表連起來，條件是當item_sk值相等且item_id等于一個固定值。DeltaLake數據湖基礎介紹(商業版)51如果不開啟DFP,從上圖可以看出，查詢會先對store_sales進行全濾后的item表的行進行join,雖然結果僅有4中的86多億條數據。DeltaLake數據湖基礎介紹(商業版)52萬條store_sales中的數據，比未啟用DFP時減少了近99%。從結果上看，啟動DFP后，該條查詢實現了10倍的性能提升。針對該特性在TPC-DS上進行測試(見下圖),測試發現啟用DFP后，TPC-DS的查詢速度達到4.5倍到8倍的提升。且通過數據布局優化，建立輔助數據結構減少對非緩存數據讀取的I/O,實現了01.什么是克隆兩種克隆方式：shallow(淺克隆),deep(深克隆)。1)深克隆語句2)深克隆的特性⑩與源表相比，克隆表有獨立的歷史記錄；⑩在克隆過程中、或之后發生的對源表的任何更改，都不會反映在克隆表中；1)淺克隆語句與深克隆語句類似，只是在SQL中加入SHALLOWCLONE語句；在Python和Scala中2CREATETABLE3.forName(4.clone("path/to/copy",isShallow=tr2)淺克隆的特性⑩淺克隆不是自包含的，即自身不是數據源，如果源文件數據被刪除，則淺克隆數據可能會不可用；⑩淺克隆不復制流事務或COPYINTO相關的元數據；4.克隆的使用場景克隆的適用場景有很多，比如：數據存儲、短期實驗、數據分享和災難恢復，其中除了短期實驗使用淺克隆，其它場景都需要使用深克隆。如何使用DeltaLake構建批流一體數據倉庫55六、如何使用DeltaLake構建批流一體數據倉庫作者：李元健、馮加亮相信大家在構建數倉處理數據方面都很有經驗，而產業界也耗費了大量的資源來構建相關的系統。分別以不同的處理形式為用戶提供服務。那么我們期望的理想的系統是什么樣的?①更一體化或更加聚焦，讓更專業的人干更專業的事情⑩有同時處理流式和批量的能力如何使用DeltaLake構建批流一體數據倉庫56理想與現實理想與現實低質、不可靠數據2.想要解決的問題1)歷史查詢如何使用DeltaLake構建批流一體數據倉庫57spark第一條處理流比較簡單，比如通過ApachSpark直接使用StreamingAnalytics打通實時與此同時，需要離線流時，歷史查詢可以使用Lambda架構對應的方式。ApachSpark提供了很好的抽象設計，我們可以通過一種代碼或API來完成流和實時的入架構設計。通過歷史數據的查詢，我們可以進一步使用Spark進行SQL分析，以及用SparkSQL的作業的形式來產生AI技術的能力。2)數據校驗A-arch①接下來我們需要面對的第一個問題就是數據的校驗。我們的流式數據和批量數據，假設以Lambda架構的形式存在時，如何確認我們在某一如何使用DeltaLake構建批流一體數據倉庫58個時間點查出來的數據是對的?到底流式的數據和批量的數據差多少?我們的批量數據什么時候該與流式數據進行同步?所以Lambda架構還需要引入Validation,這需要我們予以確認。尤其是像報表系統面向用戶的這種精確的數據分析系統，Validation這一步驟不可或缺。因此，也許我們需要一支旁支來解決流式和批量之間的同步問題，以及對應的驗證問題。3)數據修復spork假設如上問題解得到了解決，在系統上了一段時間我們會發現，如我們對應的某個Partitioned數據出了問題，當天的臟數據在若干天之后需要修正。此時我們需要怎么辦?通常，我們需要停掉線上的查詢后再修復數據，修復完數據后重新恢復線上的任務。如此折騰的過程，實際無形的給系統架構又增加了一個修復以及過去版本回復的能力。因此，Reprocessing誕生了。4)數據更新如何使用DeltaLake構建批流一體數據倉庫59A假設解決完了Reprocessing問題，我們在AI和Reporting最終的出口端，可以看到有新的一系列的需求。比如有一天業務部門或者上級部門、合作部門提出能否Schema5)理想中的DeltaLake如何使用DeltaLake構建批流一體數據倉庫601.DeltaLake具備的能力→→→→→→4.具備在線處理歷史數據的能力3)歷史數據以及臟數據的回滾。我們需要有TimeTravel的能力來回溯到某一個時間點5)可以在不阻斷下游作業的前提下處理遲到的數據，可以直接入表。如何使用DeltaLake構建批流一體數據倉庫61以上5點完全解決之后，我們就可以用DeltaLake來替代Lambda架構，或者說我們一系列批流分制的架構設計可以使用DeltaLake架構?；诨贒eltaLake的架構設計統一批量、流式的持續數據流模型什么是基于DelatLake的架構設計?DelatLake的架構設計中一系列的元數據或者最低的級別就是表。可以將我們的數據一層一層的分成基礎數據表，中間數據表以及最終的高質量數據表。所有的一切只需要關注的就是表的上游和下游，它們之間的依賴關系是不是變得更加的簡單和干凈。我們只需要關注業務層面的數據組織，所以DelatLake是統一批量、流式的持續數據流的模型。以下通過Demo的形式演示如何在Databricks數據洞察里搭建批流一體數據倉庫的操作，解決生產環境的問題。Demo演示視頻：/live/248826七、使用DDI+Confluent進行實時數據采集入湖和分析本文將介紹使用Python腳本周期性地向KafkaBrokers發ComectksqiDBumlflow上圖最左側為lOT設備和移動應用，負責采集設備或者應用的運行數據，發布至Kafka本示例中模擬實時數據的數據集為紐約市出租車數據，存儲在CSv文件內。數據中第一列下車的經緯度以及乘客的數量。Q進入Confluent管控臺，Jinxi-DDl-test為本次使用的confluent集群。使用DDI+Confluent進行實時數據采集入湖和分析643登錄ControlCenter,可以查看本集群的詳細信息，比如broker數量、broker產生和消費的吞吐量、topic數量、partition數量等信息。創建一個用于接收數據的topic,名為nyc_taxi_data,partition數量設置為3,與broker數量一致。92Python腳本里需要提供集群ID,ControlCenter的用戶名、密碼以及topic名稱。創建一個名為conf的dict,指定bootstrapserver的地址，此地址需要根據ClusterlD拼接而成，其他都為樣板代碼。生成一個producer,其來源于confluentKafka庫，因此用戶需要安裝此庫。打開train文件，讀取CSV文件里的數據并發送到confluent集群。G進入ControlCenter,可以看到已經開始生產消息。使用DDI+Confluent進行實時數據采集入湖和分析66Aipoa9.o/Pwt(ny:20186-51:2La26n“,"fareim4,5.,“aickupatetta3844-2517:24;ure,auranlagjita進入topic,點擊message,可以直接跳轉到offset0,即最開始的消息，可以查看消息至此，數據的采集和發布鏈路已經打通，下一步需要到DDINotebook連接到Kafkatreaming-Data-Proces使用DDI+Confluent進行實時數據采集入湖和分析67--這兩個屬性是Databricks數據洞察相對開源版deltalake特有的，Spark流式數據入湖時，--會產生大量的小文件，過多的小文件使得delta表的元數據變得不可擴展，且使得查詢速度下降--我們提供的這兩個特性可以自動合并數據入湖過程中產生的小文件，避免出現查詢性能下降和元數據擴展性的問題(delta.autoOptimizerodelta.autoOptimize.autoCompact此外還設置了表的屬性，主要與DDI引擎的特性相關。第一個frompyspark.sql.functionsinportfron_json,col,decode,explodeconfluent_serverf"rb-{confcontrol_center_password-os.environ[L'pcheckpoint_Locatibn="oss://databricks-delta-demo/taxi_data/chectaxi_data_delta_lake="oss://databricks-delta-demo/taxi_data/data".option("kafka.bootstrap.servers".option("startingoffs.option("kafka.security.pr.option("kafka.saslf"""org.apache,mon,security.plain.PlainLoginMorequiredusername="{control_center_username}"password-"{control_center_passwschema=(StructType().add('key',TimestampT.add('fare_amount',FLoatType().add('pickup_datetime',TimestampType()).add('pickup_latitude.add('dropoff_longitude',.add('passenger_count',IntegerType()))lines-(lines.withColumn('data',fron_json(col('value').cast('string'),schemquery=(lines.writeStreon.fornat('deltaquery=(lines.writeStreon.fornat('delta.option('path',taxi_data_d集群進行消費。消費邏輯與發送邏輯較為相近，需要給記錄消費Kafka數據時的offset,如果作業中斷或異常停止，重啟后可以從對應使用DDI+Confluent進行實時數據采集入湖和分析68在value里，因此需要創建一列data,負責從value里取值。value為二進制形式，需◎設置◎設置關于LTASKWScuUxPCq誓子用戶oss//databricks-deta-demo/taxi_data/check類型/大小最后修改時間Select若干數據進行查看。在OSS的browser里可以看到注入的數據以及過程中產生m三使用DDI+Confluent進行實時數據采集入從交易量的走勢圖可以看出，3、4、5、6月的交易量較高，8月達到低谷。selectdayofweek('pickup_datetime)asgroupbydayofweek('43254325Took3sec.LastupdatedbyJinxiat統計在每周內從周一到周日的交易量變化，如上圖。fook3sec.Lastupdatedby統計一天中每個時間段的交易量，如上圖?？梢钥闯雒刻熘形绾拖挛鐬榻灰琢康牡凸?。統計每個時間段打車的費用，如上圖。可以看出中午打車費用較高。因此可以得出結論：由于打車價格攀升導致了交易量萎靡。.統計每個年份打車費用的均值，如上圖左，顯示為打車費用連年上升。再統計每個年份的交易量，如上圖右，可以得出結論：交易量和打車費用成反比關系。--統計交易金額的分布where'fare_amount>0and'fare_amoungroupbycastC'fare_amount'3count(1)132,147統計交易金額的分布，如上圖?？梢钥闯觯~約市打車價格主要分布在3-20美元之間。在以上查詢分析的過程中，流式處理作業一直處于運行狀態，與批式作業沒有沖突，可以并行運行。費KafkaBrokers中的數據，實現了實時數據入湖，并且使用DDI的Zeppelin基于入湖的實時數據做一些簡單分析。使用Databricks進行零售業需求預測的應作者：李錦桂本文從零售業需求預測痛點、商店商品模型預測的實踐演示，介紹Databricks如何助力零售商進行需求、庫存預測，實現成本把控和營收增長。首先，需求預測對零售商至關重要。如果商店的商品過多，貨架和倉庫的空間緊張，產品容易過期，財務資源被庫存束縛。零售商無法利用制造商帶來的新機會，從而錯過消費模式的轉變。由于商店內商品過少，客戶無法從上商店內買到需要的商品。不但會造成零售商的收入損失，而且隨著時間的推移，消費者的失望情緒，會驅使消費者轉向競爭對手。綜上所述，預測消費者需求的準確性和及時性，對零售商非常重要。下面我們使用零售數據模擬如何使用DDI的notebook和Facebookprophet來對消費者的需求進行建模和預測。現在我們需要的數據已經上傳到了OSS的Bucket里面，接下來，開始對消費者的需求進行建模和預測。當數據上傳到OSS上之后，可以在DDI的Notebook上對數據進行分析和建模。本次使用的數據集是2012年到2017年，10個商店中的50商品銷售數據。數據包含四列。第一列是日期；第二列是商店的ID(1-10);第三列是商品的ID(1-50);第四列是當日商品的銷售量。實驗目的是預測未來三個月，這些商品在各個商店的銷量，對商店未來的庫存備貨提供指導。使用Databricks進行零售業需求預測的應用實踐73spark.shuffle.service.enaspark.dynamicAllocation.enabledspark.dynamicAllocation.minExecut[train_data_path="oss://time-series-forcTook0sec.LastupdatedbyJinxi(train_df)使用Databricks進行零售業需求預測的應用實踐74讀取DataFrame之后，通過熟悉的SQL語句對數據進行分析，可以使用dataframe的createOrReplaceTempView方法，創建一個臨時的視圖。杰201420152016Took1sec.LastupdatedbyJinxiatApril102022,11:56:04PM.創建view之后，對dataframe中的數據進行分析。首先分析銷售數據隨年份的走勢。從圖表可以看出，在過去幾年，商店的銷售額穩步增長，總體呈現線性增長的趨勢。在預測下一年的銷售額時，可以參考過去幾年的增長率。3.基于DDI建立預測單個商店-商品模型與此同時，商品銷售往往有很強的季節性，特別是服裝行業。T恤在夏季的銷售額肯定高于羽絨服的銷售額。因此，在預測商品的銷售額時，季節性是不可忽略的因素。2013-01-012012016-05-012017-03-012017-Took1sec.LastupdatedbyJi如上圖所示，從2013年到2017年，商品銷量不斷上漲。一年之中，商品的銷售額呈現很強的周期性。在12月或1月時，商品銷量到達波谷，隨著月份不斷攀升，7月銷量到達波峰。所以在進行建模時，月份是很