表流讀取和寫入

限製輸入的速度

以下選項可以控製micro-batches:

• maxFilesPerTrigger:有多少新文件被認為是在每個micro-batch。默認值是1000。

• maxBytesPerTrigger在每個micro-batch:多少數據被處理。這個選項設置一個“軟馬克斯”,這意味著一個批處理過程大約這個過程的數據量,可能超過極限為了使流查詢前進情況下的最小輸入單位超過這個極限。如果你使用Trigger.Once為你流,此選項將被忽略。這不是默認設置。

如果你使用maxBytesPerTrigger結合maxFilesPerTrigger,micro-batch過程數據,直到maxFilesPerTrigger或maxBytesPerTrigger達到極限。

忽略更新和刪除

結構化流不處理不是一個附加的輸入,將拋出一個異常如果發生任何修改在桌子上被用作源。有兩個主要的策略來處理變化,不能自動下遊傳播:

• 您可以刪除輸出從一開始就和檢查點和重啟流。

• 你可以設置這兩個選擇:

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

  • ignoreChanges:如果文件必須被重寫處理文檔更新源表中的數據修改操作,比如更新,合並成,刪除(分區),或覆蓋。不變行可能仍在釋放,因此你的下遊消費者應該能夠處理重複。刪除不是下遊傳播。ignoreChanges包容ignoreDeletes。因此如果你使用ignoreChanges,你流將不會被刪除或更新源表。

火花。readStream。格式(“δ”)。選項(“ignoreDeletes”,“真正的”)。負載(“/ tmp /δ/ user_events”)

火花。readStream。格式(“δ”)。選項(“ignoreChanges”,“真正的”)。負載(“/ tmp /δ/ user_events”)

指定初始位置

您可以使用以下選項指定的起點三角洲湖流源沒有處理整個表。

• startingVersion:從三角洲湖版本。所有表變化從這個版本(包容)將讀取流源。你可以獲取提交的版本版本列的描述曆史命令的輸出。

  在磚運行時的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”)

沒有數據被刪除過程初始快照

使用增量表作為流源時,數據的查詢過程首先出現在桌子上。三角洲表在這個版本被稱為初始快照。默認情況下,三角洲表的數據文件處理基於文件最後修改。然而,最後修改時間並不一定代表記錄事件的時間順序。

在有狀態流查詢定義水印,處理文件的修改時間會導致記錄處理錯誤的訂單。這可能導致晚期事件的記錄刪除水印。

你可以避免數據下降問題通過支持下列選項:

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

啟用事件時間順序後,初始快照數據的事件時間範圍分為桶。每個微一桶的批處理流程過濾數據的時間範圍內。maxFilesPerTrigger和maxBytesPerTrigger配置選項仍適用於控製microbatch大小,但隻在一個近似的方法由於處理的本質。

下圖顯示了這個過程:

明顯的這一特性的信息:

• 數據下降問題隻有當初始增量快照的狀態在默認順序流查詢處理。

• 你不能改變withEventTimeOrder一旦流查詢開始時初始快照還是正在處理。重新啟動,withEventTimeOrder改變,你需要刪除檢查點。

• 如果您正在運行一個啟用了withEventTimeOrder流查詢,你不能降級到DBR版本不支持這個功能,直到初始快照處理完成。如果你需要降級,你可以等待初始快照完成,查詢或刪除檢查點和重啟。

• 不支持這個功能在以下常見場景:

  • 事件時間列是一個生成的列有non-projectionδ源和水印之間的轉換。

  • 有一個水印,不止一個δ源在流查詢。

• 啟用事件時間順序後,三角洲初始快照處理的性能可能較慢。

• 每一批微掃描初始快照來過濾數據對應的事件時間範圍內。加快過濾操作,建議使用一個δ源列事件時間,以便數據不可以應用(檢查數據不在適當的時候)。另外,表分區沿著事件時間列可以進一步加快處理。您可以檢查火花UI,看看有多少δ特定微批處理文件掃描。

火花。readStream。格式(“δ”)。選項(“withEventTimeOrder”,“真正的”)。負載(“/ tmp /δ/ user_events”)。withWatermark(“event_time”,“10秒”)

指標

你可以找到的字節數和文件數量有待處理流媒體查詢流程隨著numBytesOutstanding和numFilesOutstanding指標。如果您正在運行流在一個筆記本,你可以看到這些度量標準下原始數據選項卡中流儀表板查詢進展:

{“源”:({“描述”:“DeltaSource(文件/道路/ /源):“,“指標”:{“numBytesOutstanding”:“3456”,“numFilesOutstanding”:“8”},}]}

Append模式

默認情況下,流在附加模式下運行,向表添加新記錄。

您可以使用路徑的方法:

事件。writeStream。格式(“δ”)。outputMode(“添加”)。選項(“checkpointLocation”,“/ tmp /δ/ _checkpoints /”)。開始(“/δ/事件”)

事件。writeStream。格式(“δ”)。outputMode(“添加”)。選項(“checkpointLocation”,“/ tmp /δ/事件/ _checkpoints /”)。開始(“/ tmp /δ/事件”)進口io。δ。值得一提的。_事件。writeStream。outputMode(“添加”)。選項(“checkpointLocation”,“/ tmp /δ/事件/ _checkpoints /”)。δ(“/ tmp /δ/事件”)

或者是toTable方法在火花3.1和更高版本(磚運行時的8.3及以上),如下所示。(在火花之前版本3.1(磚運行時8.2及以下),使用表方法。)

事件。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”)

前麵的例子不斷更新一個表,其中包含客戶總數量的事件。

更寬鬆的延遲需求的應用程序,你可以節省計算資源與一次性觸發器。使用這些更新總結聚合表在給定的時間表,隻處理新數據到達自上次更新。

例子

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)。保存(…)/ /位置1batchDF。寫。格式(…)。選項(“txnVersion”,batchId)。選項(“txnAppId”,appId)。保存(…)/ /位置2}