引入Databrick攝取：從不同來源輕鬆高效地將數據攝取到Delta Lake
我們很高興在公共預覽中推出一項新功能–自動加載程序-和一組合作夥伴集成，該集成允許Databricks用戶從各種數據源向Delta Lake增量攝取數據。Auto Loader是Apache Spark的優化雲文件源，可在新數據到達時從雲存儲連續有效地加載數據。合作伙伴集成的數據攝取網絡使您可以將數百個數據源中的數據直接攝取到Delta Lake中。
彙集所有數據
組織在各種數據源中都有大量的孤立信息。這些可能會有所不同，從數據庫（例如Oracle，MySQL，Postgres等）到產品應用程序（Salesforce，Marketo，HubSpot等）。大量的分析用例需要來自這些不同數據源的數據來生成有意義的報告和預測。例如，完整的渠道分析報告將需要來自各種來源的信息，這些信息從樞紐中的潛在客戶信息到Postgres數據庫中的產品註冊事件。
僅將所有數據集中在數據倉庫中是一種反模式，因為Python / R庫中的機器學習框架將無法有效訪問倉庫中的數據。由於您的分析用例涵蓋從構建簡單的SQL報告到更高級的機器學習預測的範圍，因此至關重要的是，您必須使用來自所有數據源的數據以開放格式構建一箇中央數據湖，並使其可用於各種用例。
自從我們去年將Delta Lake開放源代碼以來，成千上萬的組織以一種開放的格式構建中央數據湖，比以前更加可靠和高效。Delta Lake on Databricks提供ACID事務和高效的索引編制，這對於為各種訪問模式（從BI工具中的即席SQL查詢到計劃的脫機培訓工作）公開數據至關重要。我們稱這種為開放式格式的數據構建集中，可靠和有效的單一事實來源的模式稱為“ The Lakehouse”，用於從BI到ML的解耦存儲和計算的用例。
圖1. Delta Lake的通用數據流。數據被加載到攝取表中，在連續的表中進行精煉，然後用於ML和BI用例。
建造湖舍的一項關鍵挑戰是將來自各種來源的所有數據整合在一起。根據您的數據旅程，數據團隊有兩種常見方案：
- 從第三方獲取數據：通常，您在各種內部數據源（從Hubspot到Postgres數據庫）中都有有價值的用戶數據。您需要為每個連接器編寫專用的連接器，以從源中提取數據並將其存儲在Delta Lake中。
- 從雲存儲中提取數據：您已經具有一種將數據從源中拉到雲存儲中的機制。當新數據到達雲存儲時，您需要識別此新數據並將其加載到Delta Lake中進行進一步處理。
來自第三方的數據提取
從內部數據源攝取數據需要為每個數據源編寫專用的連接器。這可能是在時間和精力上的巨大投資，需要使用源API來構建連接器並將源模式映射到Delta Lake的模式功能。此外，隨著源的API和架構的發展，您還需要維護這些連接器。維護問題與您擁有的每個其他數據源都更加複雜。
為了使您的用戶更輕鬆地訪問Delta Lake中的所有數據，我們現在與一組數據提取產品合作。這個數據攝取合作伙伴網絡已與Databricks建立了本機集成，以將數據直接攝取並存儲在Delta Lake中的雲存儲中。這可以幫助您的數據科學家和分析人員輕鬆地開始使用來自各種來源的數據。
Azure Databricks客戶已經從與Azure Data Factory的集成中受益，可以將各種來源的數據提取到雲存儲中。我們很高興地宣佈一組新的合作伙伴- Fivetran，Qlik，Infoworks，StreamSets和Syncsort公司 -幫助用戶採集數據來自多種來源。我們還將通過合作伙伴Informatica，Segment和Stitch的更多集成來擴展合作伙伴的數據獲取網絡。
圖2.數據攝取合作伙伴和一些流行的數據源的生態系統，您可以通過這些合作伙伴產品將數據提取到Delta Lake。
從雲存儲中提取數據
ETL工作負載中的常見工作流是：在將新數據降落到雲Blob存儲上時對其進行增量處理，使其準備好進行分析。但是，很難以低成本，低延遲和最少的DevOps工作量一次準確地從雲Blob存儲中連續加載數據。
數據進入Delta表後，藉助Delta Lake的ACID事務，可以可靠地讀取數據。要從Delta表中流式傳輸數據，您可以使用Delta源（Azure | AWS），該源利用表的事務日誌快速識別添加的新文件。
但是，主要瓶頸在於將存儲在雲存儲中的原始文件加載到Delta表中。基於天真的文件的流媒體源（Azure | AWS）通過重複列出雲目錄並跟蹤看到了哪些文件來標識新文件。由於重複列出文件，將越來越多的文件添加到目錄中時，成本和延遲都會迅速增加。為了解決此問題，數據團隊通常會採用以下解決方法之一：
- 端到端的高數據延遲：儘管數據每隔幾分鐘到達一次，但是您可以將數據批處理到一個目錄中，然後按計劃進行處理。使用基於日或小時的分區目錄是一種常用技術。這延長了SLA，使數據可供下游使用者使用。
- 手動DevOps方法：要使SLA保持較低水平，您可以選擇利用雲通知服務和消息隊列服務來通知新文件何時到達消息隊列，然後處理新文件。這種方法不僅涉及所需雲服務的手動設置過程，而且在有多個需要加載數據的ETL作業時，管理起來也會很快變得複雜。此外，重新處理目錄中的現有文件還需要手動列出文件並處理它們（除了雲通知設置外），從而增加了設置的複雜性。
Auto Loader是一種優化的文件源，它克服了上述所有限制，併為數據團隊提供了一種無縫方式，以最小的DevOps努力以低成本和低延遲加載原始數據。您只需要提供源目錄路徑並開始流作業即可。新的結構化流媒體源稱為“ cloudFiles”，將自動設置文件通知服務，該服務從輸入目錄訂閱文件事件並在新文件到達時對其進行處理，並且還可以選擇處理該目錄中的現有文件。
圖3.以低SLA實現一次準確的數據提取需要手動設置多個雲服務。Auto Loader開箱即用地處理所有這些複雜性。
使用自動加載器的主要好處是：
- 無文件狀態管理：源將新文件降落到雲存儲上時，源將對其進行增量處理。您不需要管理有關到達哪些文件的任何狀態信息。
- 可擴展：源將通過利用雲服務和RocksDB有效地跟蹤到達的新文件，而不必列出目錄中的所有文件。即使目錄中包含數百萬個文件，此方法也可擴展。
- 易於使用：源將自動設置增量處理文件所需的通知和消息隊列服務。您無需進行任何設置。
使用自動加載程序流式加載
您可以利用Apache Spark熟悉的負載API 開始對流作業進行最少的代碼更改：
<code>spark.readStream.format（“ cloudFiles ”）
.option（“ cloudFiles.format”，“ json”）
.load（“ / input / path”）
/<code>
使用自動加載程序進行計劃的批量加載
如果您每隔幾個小時才收到一次數據，則仍可以使用結構化流的Trigger.Once模式在計劃的作業中利用自動加載程序。
<code>val df = spark.readStream.format（“ cloudFiles”）
.option（“ cloudFiles.format”，“ json”）
.load（“ / input / path”）
df.writeStream.trigger（Trigger.Once）
.format（“ delta”）
.start（“ /輸出/路徑”）
/<code>
您可以使用Databricks Jobs Scheduler（Azure | AWS）將上述代碼安排為每小時或每天按計劃運行，以增量方式加載新數據。使用上述方法，您無需擔心遲到的數據方案。
使用COPY命令計劃的批量加載
喜歡使用聲明性語法的用戶可以使用SQL COPY命令按計劃將數據加載到Delta Lake中。COPY命令是冪等的，因此在發生故障時可以安全地重新運行。該命令會自動忽略以前加載的文件，並保證一次語義準確。這使數據團隊可以輕鬆構建健壯的數據管道。
該命令的語法如下所示。有關更多詳細信息，請參閱有關COPY命令的文檔（Azure | AWS）。
<code>COPY INTO tableIdentifier
來自{ （選擇identifierList FROM位置）}
FILEFORMAT = {CSV | JSON | AVRO | ORC | 派克}
[文件=（''[，''] [，...]）]
[PATTERN ='']
[FORMAT_OPTIONS（'dataSourceReaderOption'='value'，...）]
[COPY_OPTIONS（'force'= {'false'，'true'}）]
/<code>
圖4.使用新功能將數據導入Delta Lake。使用Auto Loader進行流式加載可確保一次準確地提取數據。可以使用COPY命令重試批量加載。
數據提取功能入門
對於機器學習和業務分析用例成功而言，將所有數據放入數據湖至關重要，這對每個組織而言都是一項艱鉅的任務。我們很高興介紹Auto Loader和合作夥伴集成功能，以幫助我們成千上萬的用戶在構建高效數據湖的過程中獲得幫助。這些功能現在可以作為預覽。我們的文檔提供了有關如何開始合作伙伴集成（Azure | AWS），自動加載程序（Azure | AWS）和複製命令（Azure | AWS）的更多信息，以開始將數據加載到Delta Lake。
閱讀更多 發家致富全靠它 的文章
