結構化流寫入Azure突觸

Azure突觸連接器提供高效和可擴展的結構化流寫支持Azure突觸與批處理提供了一致的用戶體驗並使用寫道複製對於大型磚集群之間的數據傳輸和Azure突觸實例。

結構化流支持磚和突觸之間提供了簡單的語義配置增量ETL作業。模型用於加載數據從磚到突觸介紹延遲,可能不滿足SLA要求實時工作負載。看到在Azure突觸分析查詢數據

支持輸出流寫入突觸模式

Azure突觸連接器支持附加完整的輸出模式記錄附加和聚合。為更多的細節在輸出模式和兼容性矩陣,看到結構化流導

突觸容錯語義

默認情況下,Azure突觸流提供了端到端僅一次保證寫數據到Azure突觸表可靠地跟蹤進度查詢的使用檢查點位置DBFS、檢查點表在Azure突觸,和鎖定機製,以確保流可以處理任何類型的失敗,重試,和查詢將重新啟動。

可選地,您可以選擇限製較少“至少一次”語義Azure突觸流設置spark.databricks.sqldw.streaming.exactlyOnce.enabled選項,在這種情況下,數據重複的事件可能發生間歇性連接失敗Azure突觸查詢或意外終止。

結構化流語法寫作Azure突觸

下麵的代碼示例演示流寫入突觸在Scala和Python使用結構化流:

/ /設置Blob存儲帳戶訪問關鍵會話配置在筆記本上。火花相依(“fs.azure.account.key < your-storage-account-name > .dfs.core.windows.net”。,“< your-storage-account-access-key >”)/ /準備流源;這可能是卡夫卡或一個簡單的流率。瓦爾df:DataFrame=火花readStream格式(“速度”)選項(“rowsPerSecond”,“100000”)選項(“numPartitions”,“16”)負載()/ /應用一些轉換到數據然後使用/ /結構化流API,不斷將數據寫入一個表在Azure突觸。dfwriteStream格式(“com.databricks.spark.sqldw”)選項(“url”," jdbc::狀態"置疑" / / < the-rest-of-the-connection-string >”)選項(“tempDir”,“abfss: / / < your-container-name > @ < your-storage-account-name >.dfs.core.windows.net/ < your-directory-name >”)選項(“forwardSparkAzureStorageCredentials”,“真正的”)選項(“數據表”,“< your-table-name >”)選項(“checkpointLocation”,“/ tmp_checkpoint_location”)開始()
#設置Blob存儲帳戶訪問關鍵會話配置在筆記本上。火花相依(“fs.azure.account.key < your-storage-account-name > .dfs.core.windows.net”。,“< your-storage-account-access-key >”)#準備流源;這可能是卡夫卡或一個簡單的流率。df=火花readStream\格式(“速度”)\選項(“rowsPerSecond”,“100000”)\選項(“numPartitions”,“16”)\負載()然後使用#應用一些轉換數據#結構化流API,不斷用Azure突觸寫數據表。dfwriteStream\格式(“com.databricks.spark.sqldw”)\選項(“url”," jdbc::狀態"置疑" / / < the-rest-of-the-connection-string >”)\選項(“tempDir”,“abfss: / / < your-container-name > @ < your-storage-account-name >.dfs.core.windows.net/ < your-directory-name >”)\選項(“forwardSparkAzureStorageCredentials”,“真正的”)\選項(“數據表”,“< your-table-name >”)\選項(“checkpointLocation”,“/ tmp_checkpoint_location”)\開始()

配置的完整列表,請參閱在Azure突觸分析查詢數據

突觸流表管理檢查點

Azure突觸連接器刪除流檢查點表時創建新的流媒體查詢開始。這種行為是一致的checkpointLocation通常指定對象存儲。磚建議您定期刪除檢查點的表進行查詢,在未來不會運行。

默認情況下,所有檢查點表名稱<前綴> _ < query-id >,在那裏<前綴>是一個可配置的前綴與默認值databricks_streaming_checkpointquery_id是一個流媒體查詢ID_角色刪除。

尋找過期或刪除所有檢查點表流查詢,運行查詢:

選擇*sys在哪裏的名字就像“databricks_streaming_checkpoint %”

您可以配置前綴與火花SQL配置選項spark.databricks.sqldw.streaming.exactlyOnce.checkpointTableNamePrefix

磚突觸連接器流選擇參考

選項火花SQL支持以下選項中提供流媒體除了批處理選項:

參數

要求

默認的

筆記

checkpointLocation

是的

沒有默認的

位置DBFS,將使用結構化流寫元數據和檢查點的信息。看到從故障中恢複檢查點在結構化流編程指南。

numStreamingTempDirsToKeep

沒有

0

表明有多少(最新)臨時目錄保持定期清理流微批次。當設置為0目錄刪除後立即觸發微批提交,另有規定的最新微批量保存和刪除目錄。使用1禁用定期清理。

請注意

checkpointLocationnumStreamingTempDirsToKeep隻有相關的流媒體寫道磚在Azure突觸一個新表。