事件流中心提高處理速度

[Jreco]

你好,

我工作與活動中心和數據磚實時處理和豐富的數據。

做一個“簡單的”測試,我得到一些奇怪的值(輸入速率和處理速率),我覺得我失去數據:

如果你可以看到,有一個峰5 k記錄但從未加工後的5分鍾。

我使用的腳本:

conf配置[' eventhubs = {}。connectionString '] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt (connectionString_bb_stream) conf [' eventhubs。consumerGroup '] = ' adb_jr_tesst配置[' maxEventsPerTrigger '] = 350000年的會議(“maxRatePerPartition”) = ' 350000 '參看[' setStartingPosition '] = sc._jvm.org.apache.spark.eventhubs.EventPosition.fromEndOfStream df =(火花。readStream .format (eventhubs) .options (* * conf) .load json_df = df ())。withColumn(“身體”,from_json(坳(“身體”).cast(字符串),jsonSchema)) Final_df = json_df。選擇([“sequenceNumber”、“抵消”、“enqueuedTime”,上校(“。*”)])Final_df = Final_df。withColumn用戶(“關鍵”,sha2 (concat(坳(EquipmentId)坳(TagId)坳(“時間戳”)),256))Final_df.display ()

你能幫我理解為什麼我“失去”數據或如何可以改善這個過程嗎?

我使用的集群:

我認為是一個集群配置的問題,但是我不知道怎麼解決。

謝謝你的幫助,夥計們!

[Kaniz]

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

[Jreco]

嗨kaniz,
謝謝你的回複。

[Jreco]

嗨@Kaniz Fatma,很抱歉打擾你,

你能看看這個嗎?

謝謝你的幫助!

[werners1]

好的,我唯一注意到你有設置一個終止時間為流媒體沒有必要(如果你在做實時)。

我也注意到你不設置一個檢查點的位置,你可能會考慮的事情。

你也可以嚐試刪除maxEvent maxRate配置。

文檔的片段:

以下是推薦的任務配置的細節。

• 集群:設置這個總是使用一個新的集群和使用最新的火花版本(或至少2.1版本)。查詢開始引發2.1及以上的查詢和火花版本升級後可恢複。
• 警報:設置這個如果你希望電子郵件通知失敗。
• 時間表:不設置一個時間表。
• 超時:不設置一個超時。流媒體查詢無限期地為一個長的時間。
• 最大並發運行:設置為1。同時隻能有一個實例,每個查詢活躍。
• 重試:設置為無限的。

https://docs.www.eheci.com/spark/latest/structured-streaming/production.html

https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html