Databricks數(shù)據(jù)洞察從入門到實戰(zhàn)

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