實現穩定的火花的結構化流媒體的應用程序

[fsm]

嗨,夥計們,

我有一個問題。這不是關鍵,但很煩人。

我們實現了一個火花結構化流媒體應用程序。

這個應用程序將會觸發線Azure數據工廠(每8分鍾)。好的,這個設置聽起來有點奇怪,這不是真的流,同意了。但源係統不是實時的,我們想要實現一個流POC,看看工藝深有關。這是所有。

所以,這“流筆記本”跑24/7。主要是在一個穩定的方式。但有時一個運行加載到TimeoutException是這樣的:

java . util . concurrent。TimeoutException:流執行線程流(id = 26 c3f28c - 9 d17 - 486 a - 81——df418c42cd74 runId = d30a8fe8 - 3 -美聯儲- 4475 - 8233 - 4577 - b775bb19]未能阻止15000毫秒內(由spark.sql.streaming.stopTimeout指定)。看到什麼原因被流查詢線程中執行。

這個異常是明確的。這正試圖獲得加載到存儲賬戶,檢查點位置。超時,因為另一個負載沒有星期幾工作和鎖定檢查點,。一般來說這個流媒體應用程序不需要超過1 - 2分鍾。

但是一些邊界情況需要超過14分鍾和其他模塊加載,將開始在這。

司機日誌我做了一些調查,發現一個奇怪的行為到log4j(見附加log4j_snippet。日誌,時間間隔為99秒時間嗎?)

在這些邊界情況我有很多這樣的條目。

ClusterLoadAvgHelper……這是什麼意思?什麼好主意嗎?

我難過的時候,我自己的小業務邏輯和連接到SQL Azure數據庫(下沉)隻需要1 - 2分鍾。在這個邊緣情況下整個處理時間跑到14分鍾超過10分鍾ClusterLoadAvgHelper東西是必要的。

目前我也不知道為什麼我的集群運行胡作非為。

像我難過一開始並不是關鍵,我們不要錯過任何數據到SQL數據庫。但它很煩人:)。

有什麼想法就好了。

提前謝謝,

馬庫斯

數據來源:汽車加載器這樣的機製(https://docs.m.eheci.com/spark/latest/structured-streaming/auto-loader.html)

水槽:SQL Azure數據庫

設置

語言:Python3

磚運行時:8.3(包括Apache火花3.1.1,Scala 2.12)

司機/工人類型:Standard_E4ds_v4

集群模式:標準

最小工作人員:1 / Max工人:10

在- - - - - - - - -流筆記本片段

#(1)初始化流數據幀

streaming_df = spark.readStream.format \ (“cloudFiles”)

.option (“cloudFiles。形式at", extension) \

. schema \(模式)

.load (streaming_path) #定位存儲賬戶

#(2)開始流

查詢= (

streaming_df

.writeStream

.foreachBatch (process_batch_for_streaming) #沉入SQL Azure數據庫

.trigger(一旦= True)

checkpoint_path .option (“checkpointLocation”)

.start ()

)

————流筆記本片段

[Kaniz]

嗨@Markus Freischlad !我的名字叫Kaniz,我這裏的技術主持人。很高興認識你,謝謝你的問題!看看你的同行在社區中有一個回答你的問題。否則我將盡快給你回電。謝謝。

[fsm]

你好,

我已經調查了一點。目前我認為ClusterLoadAvgHelper行為不是主要的問題。這種行為隱藏這一個:

22/02/08 22:52:09警告TaskSetManager:失去了任務0.0在197663.0 (TID 4615729)階段(410年10.141.64.7執行人):com.microsoft.sqlserver.jdbc。SQLServerException:連接超時(失敗)

com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDriverError (SQLServerException.java: 234)

com.microsoft.sqlserver.jdbc.SimpleInputStream.getBytes (SimpleInputStream.java: 352)

com.microsoft.sqlserver.jdbc.DDC.convertStreamToObject (DDC.java: 796)

com.microsoft.sqlserver.jdbc.ServerDTVImpl.getValue (dtv.java: 3777)

com.microsoft.sqlserver.jdbc.DTV.getValue (dtv.java: 247)

com.microsoft.sqlserver.jdbc.Column.getValue (Column.java: 190)

com.microsoft.sqlserver.jdbc.SQLServerResultSet.getValue (SQLServerResultSet.java: 2054)

com.microsoft.sqlserver.jdbc.SQLServerResultSet.getValue (SQLServerResultSet.java: 2040)

com.microsoft.sqlserver.jdbc.SQLServerResultSet.getString (SQLServerResultSet.java: 2511)

在org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils。美元anonfun makeGetter 12美元(JdbcUtils.scala: 445)

在org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils。美元anonfun makeGetter 12美元美元改編(JdbcUtils.scala: 443)

在另一次1.美元美元org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils getnext (JdbcUtils.scala: 353)

在另一次1.美元美元org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils getnext (JdbcUtils.scala: 335)

我將做進一步檢查為SQL Azure數據庫(數據,re-org運行. .)與sql-spark-connector相結合。也許有一個流之間的衝突和數據/ reorg。

而且我將log4j配置以獲得更多細節

com.microsoft.sqlserver.jdbc。*

謝謝,

馬庫斯

[jose_gonzalez · Accepted solution]

嗨@Markus Freischlad,

你嚐試運行流工作有不同的觸發間隔?為了隔離問題。如果你使用DBR 10.1 +有一個新的觸發這個DBR版本中引入的。

新觸發“trigger.AvailableNow”。這就像觸發器。然後停止一次,哪些流程所有可用的數據查詢。然而,觸發。AvailableNow提供了更好的可伸縮性,因為可以處理多個批次的數據而不是一個。

這裏的文檔https://docs.m.eheci.com/release-notes/runtime/10.1.html triggeravailablenow-for-delta-source-str……

[匿名]

@fsm——謝謝你的額外的信息!