在Databricks上運行第一個ETL工作負載

需求

• 您已經登錄到Databricks，並且處於Data Science & Engineering工作區中。有關更多信息，請參見注冊免費試用．

• 你有創建集群權限．

步驟1:創建集群

為了進行探索性數據分析和數據工程，需要創建集群來提供執行命令所需的計算資源。

1. 點擊計算在側邊欄。

2. 在“計算池”頁麵，單擊創建集群．這將打開New Cluster頁麵。

3. 為集群指定一個惟一的名稱，其餘值保持默認狀態，然後單擊創建集群．

要了解有關Databricks集群的更多信息，請參見集群．

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

要開始在Databricks上編寫和執行交互式代碼，請創建一個筆記本。

1. 點擊創建在側欄中，然後單擊筆記本．

2. 在“創建筆記本”頁麵:

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

   • 確保默認語言設置為PythonorgydF4y2BaScala．

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

   • 點擊創建．

筆記本打開時頂部有一個空格子。

要了解關於創建和管理筆記本的更多信息，請參見管理筆記本．

步驟3:配置Auto Loader將數據攝取到Delta Lake

Databricks推薦使用自動加載程序對於增量數據攝取。當新文件到達雲對象存儲時，Auto Loader自動檢測和處理。

Databricks建議使用三角洲湖．Delta Lake是一個開源存儲層，它提供ACID事務並支持數據湖屋。Delta Lake是在Databricks中創建的表的默認格式。

要配置Auto Loader將數據攝取到Delta Lake表中，請將以下代碼複製並粘貼到筆記本中的空單元格中:

#導入函數從pyspark.sql.functions進口input_file_name#定義下麵代碼中使用的變量file_path＝“/ databricks-datasets / structured-streaming /事件”用戶名＝火花．sql（的“選擇regexp_replace (current_user (), [^ a-zA-Z0-9)”,“_”)”）．第一個（)[0］table_name＝f＂｛用戶名｝_etl_quickstart”checkpoint_path＝f“/ tmp /｛用戶名｝/ _checkpoint / etl_quickstart”#清除之前演示執行的數據火花．sql（f如果存在，刪除表｛table_name｝＂）dbutils．fs．rm（checkpoint_path，真正的）#配置Auto Loader將JSON數據導入Delta表（火花．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＝真正的）．toTable（table_name）)

/ /進口進口org．apache．火花．sql．功能{。input_file_name，current_timestamp｝進口org．apache．火花．sql．流媒體．觸發進口火花．值得一提的．＿//定義下麵代碼中使用的變量瓦爾file_path＝“/ databricks-datasets / structured-streaming /事件”瓦爾用戶名＝火花．sql（的“選擇regexp_replace (current_user (), [^ a-zA-Z0-9)”,“_”)”）.第一個．得到（0）瓦爾table_name＝s”$ {用戶名｝_etl_quickstart”瓦爾checkpoint_path＝“/ tmp /$ {用戶名｝/ _checkpoint”//清除之前演示執行的數據火花．sql（如果存在，刪除表$ {table_name｝＂）dbutils．fs．rm（checkpoint_path，真正的）//配置Auto Loader將JSON數據接收到Delta表火花．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）．toTable（table_name）

要了解更多關於自動加載器，請參見自動加載程序．

步驟4:處理數據並與之交互

筆記本逐單元執行邏輯。要在單元格中執行邏輯:

1. 要運行在上一步中完成的單元格，請選擇單元格並按下SHIFT + ENTER．

2. 要查詢剛剛創建的表，請將以下代碼複製並粘貼到一個空單元格中，然後按下SHIFT + ENTER運行計算單元。

   df＝火花．讀．表格（table_name）

   瓦爾df＝火花．讀．表格（table_name）

3. 要預覽DataFrame中的數據，請將以下代碼複製並粘貼到空單元格中，然後按下SHIFT + ENTER運行計算單元。

   顯示（df）

   顯示（df）

要了解更多關於可視化數據的交互選項的信息，請參見可視化．

第五步:安排工作

通過將Databricks筆記本添加為Databricks作業中的任務，可以將其作為生產腳本運行。在此步驟中，您將創建一個可以手動觸發的新作業。

把你的筆記本作為一項任務:

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

2. 的惟一名稱作業名．

3. 點擊手冊．

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

5. 點擊創建．

6. 在出現的窗口中，單擊現在運行．

7. 要查看作業運行結果，請單擊圖標旁邊的最後一次運行時間戳。

有關工作的更多信息，請參見工作．

額外的集成

使用Databricks了解更多關於數據工程的集成和工具:

• 連接您最喜歡的IDE

• 使用Azure數據工廠或Apache風流來管理數據管道中的依賴關係

• 在Databricks中使用dbt

• 了解Databricks命令行接口(CLI)

• 了解有關Databricks Terraform Provider的信息