表流讀寫

表作為源

當您將Delta表加載為流源並在流查詢中使用它時，查詢將處理表中出現的所有數據以及流啟動後到達的任何新數據。

您可以將路徑和表作為流加載。

火花．readStream．格式（“δ”）．負載（“/ tmp /δ/事件”）進口io．δ．值得一提的．_火花．readStream．δ（“/ tmp /δ/事件”）

或

進口io．δ．值得一提的．_火花．readStream．格式（“δ”)。表格（“事件”）

限製輸入速率

以下選項可用於控製微批次:

• maxFilesPerTrigger:每個微批處理中要考慮多少個新文件。缺省值是1000。

• maxBytesPerTrigger:每個微批處理的數據量。該選項設置了一個“軟最大值”，這意味著批處理大約這個數據量，並且可能處理超過限製的數據量，以便在最小輸入單元大於此限製的情況下使流查詢向前移動。如果你使用觸發器。一次對於流，此選項將被忽略。默認情況下沒有設置。

如果你使用maxBytesPerTrigger連同maxFilesPerTrigger時，微批處理數據直到maxFilesPerTrigger或maxBytesPerTrigger達到極限。

流Delta Lake變更數據捕獲(CDC)饋送

三角洲湖更改數據饋送記錄對Delta表的更改，包括更新和刪除。啟用後，您可以從更改數據提要進行流處理，並將邏輯寫入下遊表中處理插入、更新和刪除。雖然更改數據提要數據輸出與它所描述的Delta表略有不同，但這提供了一種解決方案，可以將增量更改傳播到數據庫中的下遊表大獎章架構．

忽略更新和刪除

結構化流不處理非追加的輸入，如果對用作源的表進行了任何修改，則拋出異常。有兩種主要策略用於處理不能自動向下遊傳播的更改:

• 您可以刪除輸出和檢查點，並從頭重新啟動流。

• 您可以設置以下兩個選項之一:

  • ignoreDeletes:忽略在分區邊界刪除數據的事務。

  • skipChangeCommits:忽略刪除或修改現有記錄的事務。skipChangeCommits包容ignoreDeletes．

火花．readStream．格式（“δ”）．選項（“ignoreDeletes”，“真正的”）．負載（“/ tmp /δ/ user_events”）

火花．readStream．格式（“δ”）．選項（“ignoreChanges”，“真正的”）．負載（“/ tmp /δ/ user_events”）

指定初始位置

您可以使用以下選項指定Delta Lake流源的起點，而無需處理整個表。

• startingVersion:從Delta Lake版本開始。從這個版本(包括)開始的所有表更改都將由流源讀取。提交版本可以從版本的列描述曆史命令的輸出。

  在Databricks Runtime 7.4及以上版本中，若要僅返回最新更改，請指定最新的．

• startingTimestamp:開始的時間戳。在時間戳或之後提交的所有表更改(包括)將被流源讀取。之一:

  • 時間戳字符串。例如,“2019 - 01 - 01 t00:00:00.000z”．

  • 日期字符串。例如,“2019-01-01”．

你不能同時設置兩個選項;你隻能使用其中一個。隻有在開始新的流查詢時才生效。如果流查詢已經啟動，並且進度已記錄在其檢查點中，則忽略這些選項。

火花．readStream．格式（“δ”）．選項（“startingVersion”，“5”）．負載（“/ tmp /δ/ user_events”）

火花．readStream．格式（“δ”）．選項（“startingTimestamp”，“2018-10-18”）．負載（“/ tmp /δ/ user_events”）

處理初始快照，不刪除數據

當使用Delta表作為流源時，查詢首先處理表中出現的所有數據。這個版本的Delta表稱為初始快照。默認情況下，Delta表的數據文件根據最後修改的文件進行處理。但是，最後一次修改時間並不一定代表記錄事件的時間順序。

在具有已定義水印的有狀態流查詢中，按修改時間處理文件可能導致記錄以錯誤的順序處理。這可能導致記錄通過水印作為後期事件刪除。

您可以通過啟用以下選項來避免數據丟失問題:

• withEventTimeOrder:初始快照是否應該按照事件時間順序進行處理。

啟用事件時間順序後，快照初始數據的事件時間範圍被劃分為多個時間桶。每個微批通過過濾時間範圍內的數據來處理一個桶。maxFilesPerTrigger和maxBytesPerTrigger配置選項仍然適用於控製微批處理大小，但由於處理的性質，隻能以近似的方式進行控製。

下圖展示了這個過程:

關於此特性的注意事項:

• 隻有在按默認順序處理有狀態流查詢的初始Delta快照時，才會發生數據丟失問題。

• 你無法改變withEventTimeOrder一旦流查詢開始，而初始快照仍在處理中。重新啟動withEventTimeOrder更改後，您需要刪除檢查點。

• 如果您正在運行啟用theventtimeorder的流查詢，在初始快照處理完成之前，您不能將其降級為不支持此功能的DBR版本。如果需要降級，可以等待初始快照完成，或者刪除檢查點並重新啟動查詢。

• 以下特殊場景不支持該特性:

  • 事件時間列是一個生成的列，並且在增量源和水印之間存在非投影轉換。

  • 在流查詢中，有一個水印具有多個增量源。

• 啟用事件時間順序後，Delta初始快照處理的性能可能會變慢。

• 每個微批處理掃描初始快照，以過濾相應事件時間範圍內的數據。為了加快篩選操作，建議使用Delta源列作為事件時間，以便應用數據跳過(檢查數據跳躍與z順序索引三角洲湖當它適用的時候)。此外，沿著事件時間列進行表分區可以進一步加快處理速度。您可以檢查Spark UI，以查看為特定的微批處理掃描了多少增量文件。

火花．readStream．格式（“δ”）．選項（“withEventTimeOrder”，“真正的”）．負載（“/ tmp /δ/ user_events”）．withWatermark（“event_time”，“10秒”）

表作為接收器

還可以使用結構化流將數據寫入Delta表。事務日誌使Delta Lake能夠保證隻進行一次處理，即使有其他流或批查詢同時對該表運行。

{“源”：［{“描述”：“DeltaSource(文件/道路/ /源):“，“指標”：{“numBytesOutstanding”：“3456”，“numFilesOutstanding”：“8”}，}］}

（事件．writeStream．格式（“δ”）．outputMode（“添加”）．選項（“checkpointLocation”，“/ tmp /δ/ _checkpoints /”）．開始（“/δ/事件”））

事件．writeStream．格式（“δ”）．outputMode（“添加”）．選項（“checkpointLocation”，“/ tmp /δ/事件/ _checkpoints /”）．開始（“/ tmp /δ/事件”）

（事件．writeStream．格式（“δ”）．outputMode（“添加”）．選項（“checkpointLocation”，“/ tmp /δ/事件/ _checkpoints /”）．toTable（“事件”））

事件．writeStream．outputMode（“添加”）．選項（“checkpointLocation”，“/ tmp /δ/事件/ _checkpoints /”）．toTable（“事件”）

（火花．readStream．格式（“δ”）．負載（“/ tmp /δ/事件”）．groupBy（“customerId”）．數（）．writeStream．格式（“δ”）．outputMode（“完整的”）．選項（“checkpointLocation”，“/ tmp /δ/ eventsByCustomer / _checkpoints /”）．開始（“/ tmp /δ/ eventsByCustomer”））

火花．readStream．格式（“δ”）．負載（“/ tmp /δ/事件”）．groupBy（“customerId”）．數（）．writeStream．格式（“δ”）．outputMode（“完整的”）．選項（“checkpointLocation”，“/ tmp /δ/ eventsByCustomer / _checkpoints /”）．開始（“/ tmp /δ/ eventsByCustomer”）

冪等表寫入foreachBatch

命令foreachBatch允許您指定在流查詢中的任意轉換後對每個微批的輸出執行的函數。這允許實現一個foreachBatch函數，該函數可以將微批輸出寫入一個或多個目標Delta表目的地。然而,foreachBatch不會使這些寫操作為冪等的，因為這些寫嚐試缺乏批處理是否正在重新執行的信息。例如，重新運行失敗的批處理可能導致重複的數據寫入。

為了解決這個問題，Delta表支持以下功能DataFrameWriter使寫函數冪等的選項:

• txnAppId:可以傳遞給每個對象的唯一字符串DataFrame寫。例如，您可以使用StreamingQuery ID作為txnAppId．

• txnVersion:作為事務版本的單調遞增的數字。

Delta表使用的組合txnAppId和txnVersion識別重複寫並忽略它們。

如果批寫操作因失敗而中斷，則使用相同的應用程序和批處理ID重新運行批處理，這將幫助運行時正確識別重複的寫操作並忽略它們。申請編號(txnAppId)可以是任何用戶生成的唯一字符串，並且不必與流ID相關。

相同的DataFrameWriter選項可用於在非流作業中實現冪等寫入。有關詳細信息,啟用跨作業的冪等寫入．

app_id=…#作為應用程序ID的唯一字符串。defwriteToDeltaLakeTableIdempotent（batch_df，batch_id）：batch_df．寫．格式（…）．選項（“txnVersion”，batch_id）．選項（“txnAppId”，app_id）．保存（…）#位置1batch_df．寫．格式（…）．選項（“txnVersion”，batch_id）．選項（“txnAppId”，app_id）．保存（…）#位置2

瓦爾appId=…//作為應用程序ID的唯一字符串。streamingDF．writeStream．foreachBatch{（batchDF：DataFrame，batchId：長）＝＞batchDF．寫．格式(…)。選項（“txnVersion”，batchId)。選項（“txnAppId”，appId)。保存(…)// location 1batchDF．寫．格式(…)。選項（“txnVersion”，batchId)。選項（“txnAppId”，appId)。保存(…)// location 2}

執行流靜態連接

您可以依賴Delta Lake的事務保證和版本控製協議來執行stream-static連接。流靜態連接使用無狀態連接將Delta表(靜態數據)的最新有效版本連接到數據流。

當Databricks在流-靜態連接中處理微批數據時，來自靜態Delta表的最新有效數據版本將與當前微批中的記錄進行連接。因為連接是無狀態的，所以您不需要配置水印，並且可以以較低的延遲處理結果。連接中使用的靜態Delta表中的數據應該是緩慢變化的。

streamingDF=火花．readStream．表格（“訂單”）staticDF=火花．讀．表格（“beplay体育app下载地址顧客”）查詢=（streamingDF．加入（staticDF，streamingDF．customer_id＝＝staticDF．id，“內心”）．writeStream．選項（“checkpointLocation”，checkpoint_path）．表格（“orders_with_customer_info”））

Upsert從流查詢使用foreachBatch

你可以使用的組合合並和foreachBatch將流查詢中的複雜upserts寫入Delta表。看到使用foreachBatch寫入任意數據接收器．

這個模式有很多應用，包括:

• 在更新模式下寫入流聚合:這比完全模式更有效率。

• 將數據庫更改流寫入Delta表:合並查詢，用於寫入更改數據可用於foreachBatch來連續地對Delta表應用更改流。

• 使用重複數據刪除將數據流寫入Delta表:用於重複數據刪除的僅插入合並查詢可用於foreachBatch使用自動重複數據刪除功能，連續向Delta表寫入數據(帶重複)。

下麵的示例演示如何在foreachBatch要完成這項任務:

//使用merge將microBatchOutputDF插入Delta表的函數defupsertToDelta（microBatchOutputDF：DataFrame，batchId：長）{//設置dataframe為view namemicroBatchOutputDF．createOrReplaceTempView（“更新”）//使用視圖名來應用MERGE//注意:你必須使用SparkSession來定義' updates '數據框架microBatchOutputDF．sparkSession．sql（”“”合並成聚合體t使用更新ON s.key = t.key當匹配時，更新集合*如果不匹配，則插入*”“”）}//將流聚合查詢的輸出寫入Delta表streamingAggregatesDF．writeStream．格式（“δ”）．foreachBatch（upsertToDelta_）．outputMode（“更新”）．開始（）

使用merge將microBatchOutputDF插入Delta表defupsertToDelta（microBatchOutputDF，batchId）：#設置數據幀為視圖名microBatchOutputDF．createOrReplaceTempView（“更新”）使用視圖名稱應用MERGE#注意:你必須使用SparkSession來定義“updates”數據框架在Databricks Runtime 10.5及以下版本中，您必須使用以下命令:# microBatchOutputDF._jdf.sparkSession () . sql(“””microBatchOutputDF．sparkSession．sql（”“”合並成聚合體t使用更新ON s.key = t.key當匹配時，更新集合*如果不匹配，則插入*”“”）#將流聚合查詢的輸出寫入Delta表（streamingAggregatesDF．writeStream．格式（“δ”）．foreachBatch（upsertToDelta）．outputMode（“更新”）．開始（））

你也可以選擇使用Delta Lake api來執行流upserts，如下例所示:

進口io．δ．表．*瓦爾deltaTable=DeltaTable．forPath（火花，“/數據/聚合物”）//使用merge將microBatchOutputDF插入Delta表的函數defupsertToDelta（microBatchOutputDF：DataFrame，batchId：長）{deltaTable．作為（“t”）．合並（microBatchOutputDF．作為（“s”)，"s.key = t.key"）．whenMatched(）.updateAll（）．whenNotMatched(）.insertAll（）．執行（）}//將流聚合查詢的輸出寫入Delta表streamingAggregatesDF．writeStream．格式（“δ”）．foreachBatch（upsertToDelta_）．outputMode（“更新”）．開始（）

從delta.tables進口*deltaTable=DeltaTable．forPath（火花，“/數據/聚合物”）使用merge將microBatchOutputDF插入Delta表defupsertToDelta（microBatchOutputDF，batchId）：（deltaTable．別名（“t”）．合並（microBatchOutputDF．別名（“s”)，"s.key = t.key"）．whenMatchedUpdateAll（）．whenNotMatchedInsertAll（）．執行（））#將流聚合查詢的輸出寫入Delta表（streamingAggregatesDF．writeStream．格式（“δ”）．foreachBatch（upsertToDelta）．outputMode（“更新”）．開始（））