教程:運行一個端到端的lakehouse分析管道

需求

• 你登錄數據磚,你在數據科學與工程工作區。有關更多信息,請參見開始:免費試用和設置。

步驟1:創建一個集群

做探索性數據分析和數據工程,創建一個集群來提供所需的計算資源執行命令。

1. 點擊計算在側邊欄。

2. 在計算頁麵,點擊創建集群。這將打開新的集群頁麵。

3. 為集群指定一個唯一的名稱。

4. 選擇單獨的節點單選按鈕。

5. 選擇單用戶從訪問模式下拉。

6. 確保你的電子郵件地址是可見的單用戶訪問字段。

7. 選擇所需的磚的運行時版本的11.1或以上,使用統一的目錄。

8. 點擊創建集群。

更多地了解磚集群,明白了集群。

步驟2:創建一個磚筆記本

開始編寫和執行互動代碼在磚上,創建一個筆記本。

1. 點擊新在側邊欄,然後單擊筆記本。

2. 在創建筆記本頁麵:

   • 為你的筆記本指定一個唯一的名稱。

   • 確保設置為默認語言Python。

   • 選擇您在步驟1中創建的集群集群下拉。

   • 點擊創建。

一個筆記本打開頂部有一個空單元。

學習更多的關於創建和管理筆記本,看到的管理筆記本。

步驟3:從外部讀寫數據位置由統一目錄管理

磚推薦使用自動加載程序增量數據的攝入。自動加載程序自動檢測和過程到達雲對象存儲新文件。

您可以使用統一目錄管理獲得外部的位置。用戶或服務主體讀文件權限在外部位置可以使用自動加載程序來攝取數據。

通常,數據將抵達一個外部位置從其他係統由於寫道。在這個演示,您可以模擬數據到達通過編寫JSON文件到一個外部位置。

將下麵的代碼複製到一個筆記本電池。替換的字符串值目錄與目錄的名稱創建目錄和使用目錄權限。替換的字符串值external_location與外部位置的路徑讀文件,寫文件,創建外部表權限。

外部位置可以被定義為一個完整的存儲容器,但往往指向一個目錄嵌套在一個容器中。

為外部位置路徑是正確的格式“s3: / / bucket名/道路/ / external_location”。

external_location=“< your_external_location >”目錄=“< your_catalog >”dbutils。fs。把(f”{external_location}/ filename.txt”,“Hello world !”,真正的)顯示(dbutils。fs。頭(f”{external_location}/ filename.txt”))dbutils。fs。rm(f”{external_location}/ filename.txt”)顯示(火花。sql(f“顯示模式{目錄}”))

執行這個細胞應該打印一行12字節寫,打印字符串“Hello world !””,顯示所有的數據庫目錄中。如果你不能得到這個電池運行,確認你在統一目錄啟用工作區和從您的工作區管理員請求適當的權限來完成本教程。

下麵的Python代碼使用您的電子郵件地址來創建一個獨特的數據庫提供的目錄中並在外部位置提供了一個獨特的存儲位置。執行這個細胞將刪除所有數據與本教程,允許您執行這個例子冪等。定義並實例化一個類,您將使用模擬批次的數據從conncted係統外部源的位置。

這段代碼複製到一個新細胞在你的筆記本和配置您的環境中執行它。

從pyspark.sql.functions進口上校#設置參數隔離在工作區和重置演示用戶名=火花。sql(的“選擇regexp_replace (current_user (), [^ a-zA-Z0-9)”,“_”)”)。第一個()(0]數據庫=f”{目錄}.e2e_lakehouse_{用戶名}_db”源=f”{external_location}/ e2e-lakehouse-source”表=f”{數據庫}.target_table”checkpoint_path=f”{external_location}/ _checkpoint / e2e-lakehouse-demo”火花。sql(f“c.username = '{用戶名}’”)火花。sql(f“設置c.database ={數據庫}”)火花。sql(f“c.source = '{源}’”)火花。sql(“數據庫,如果存在美元下降{c.database}級聯”)火花。sql(“創建數據庫{c.database}”)火花。sql(“使用美元{c.database}”)#從以前的演示執行清除數據dbutils。fs。rm(源,真正的)dbutils。fs。rm(checkpoint_path,真正的)#定義一個類加載批次的數據來源類LoadData:def__init__(自我,源):自我。源=源defget_date(自我):試一試:df=火花。讀。格式(“json”)。負載(源)除了:返回“2016-01-01”batch_date=df。selectExpr(“馬克斯(不同(日期(tpep_pickup_datetime))) + 1日”)。第一個()(0]如果batch_date。月= =3:提高異常(“源數據了”)返回batch_datedefget_batch(自我,batch_date):返回(火花。表(“samples.nyctaxi.trips”)。過濾器(上校(“tpep_pickup_datetime”)。投(“日期”)= =batch_date))defwrite_batch(自我,批處理):批處理。寫。格式(“json”)。模式(“添加”)。保存(自我。源)defland_batch(自我):batch_date=自我。get_date()批處理=自我。get_batch(batch_date)自我。write_batch(批處理)RawData=LoadData(源)

您現在可以土地一批數據,將下麵的代碼複製到一個細胞和執行它。您可以手動執行這個細胞觸發新的數據到達60倍。

RawData。land_batch()

第四步:配置自動加載程序來攝取數據統一目錄

磚建議存儲數據三角洲湖。三角洲湖是一個開源存儲層提供lakehouse ACID事務,使數據。三角洲湖是默認格式表中創建數據磚。

配置自動加載程序來攝取數據統一目錄表,下麵的代碼複製並粘貼到你的筆記本一個空單元:

#導入功能從pyspark.sql.functions進口input_file_name,current_timestamp#配置自動加載程序向三角洲攝取JSON數據表(火花。readStream。格式(“cloudFiles”)。選項(“cloudFiles.format”,“json”)。選項(“cloudFiles.schemaLocation”,checkpoint_path)。負載(file_path)。選擇(“*”,input_file_name()。別名(“source_file”),current_timestamp()。別名(“processing_time”))。writeStream。選項(“checkpointLocation”,checkpoint_path)。觸發(availableNow=真正的)。選項(“mergeSchema”,“真正的”)。toTable(表))

更多地了解汽車裝載機,明白了自動加載器是什麼?。

了解更多關於結構化流與統一目錄,看看使用統一的目錄和結構化流。

步驟5:過程和與數據交互

筆記本電腦執行邏輯細胞。使用這些步驟來執行的邏輯單元:

1. 運行的細胞在前一步完成,選擇細胞和新聞SHIFT + ENTER。

2. 查詢您剛剛創建的表,下麵的代碼複製並粘貼到一個空的細胞,然後按SHIFT + ENTER運行單元。

   df=火花。讀。表(table_name)

3. DataFrame預覽數據,下麵的代碼複製並粘貼到一個空的細胞中,然後按SHIFT + ENTER運行單元。

   顯示(df)

了解更多關於交互式可視化數據的選擇,明白了可視化。

第六步:安排一份工作

您可以運行磚筆記本生產腳本通過添加作為一個磚的工作任務。在這個步驟中,您將創建一個新的工作,你可以手動觸發。

安排你的筆記本作為一個任務:

1. 點擊時間表右邊的標題欄。

2. 輸入一個唯一的名稱作業名。

3. 點擊手冊。

4. 在集群下拉,選擇您在步驟1中創建的集群。

5. 點擊創建。

6. 在出現的窗口中,點擊現在運行。

7. 看到工作運行結果,單擊旁邊的圖標最後一次運行時間戳。

工作的更多信息,請參閱創建、運行和管理數據磚的工作。

第七步:從磚SQL查詢表

任何人的使用目錄允許在當前目錄,使用模式允許對當前模式選擇表可以查詢的權限表的內容從他們的首選磚API。

你可以切換到磚SQL UI使用上麵的角色切換器+在屏幕的左上角。選擇SQL從下拉菜單中。

你需要訪問運行SQL倉庫在磚SQL執行查詢。

本教程中前麵創建的表的名稱target_table。你可以查詢它在第一個單元格使用你提供的目錄和數據庫與論e2e_lakehouse_ < your_username >。您可以使用數據瀏覽找到您創建的數據對象。

額外的集成

了解更多關於集成數據工程與磚和工具:

• 你最喜歡的IDE連接

• 使用印度生物技術部磚

• 了解磚命令行界麵(CLI)

• 了解磚起程拓殖的提供者