Spark Structured Streaming的高效處理RunOnceTrigger是怎樣進行的

Spark Structured Streaming的高效處理RunOnceTrigger是怎樣進行的，相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

傳統意義上，當人們想到流處理時，諸如”實時”，”24*7”或者”always on”之類的詞語就會浮現在腦海中。生產中可能會遇到這種情況，數據僅僅會在固定間隔到達，比如每小時，或者每天。對于這些情況，對這些數據進行增量處理仍然是有益的。但是在集群中運行一個24*7的Streaming job就顯得有些浪費了，這時候僅僅需要每天進行少量的處理即可受益。

幸運的是，在spark 2.2版本中通過使用 Structured Streaming的Run Once trigger特性，可獲得Catalyst Optimizer帶來的好處和集群運行空閑job帶來的成本節約，這兩方面好處。

一，Structured Streaming的Triggers

在Structured Streaming中，Trigger用來指定Streaming 查詢產生結果的頻率。一旦Trigger觸發，Spark將會檢查是否有新數據可用。如果有新數據，查詢將增量的從上次觸發的地方執行。如果沒有新數據，Stream繼續睡眠，直到下次Trigger觸發。

Structured Streaming的默認行為盡可能低延遲地運行，trigger會在上次trigger觸發結束之后立即運行。針對一些有低延遲要求的使用案例，Structured Streaming支持ProcessingTime trigger，也即將會用用戶提供的時間間隔，例如每分鐘，去觸發一次查詢。

這雖然很好，但是也免不了24*7運行。相反，RunOnce Trigger僅僅會執行一次查詢，然后停止查詢。

Trigger在你啟動Streams的時候指定。

import org.apache.spark.sql.streaming.Trigger

    // Load your Streaming DataFrame
    val sdf = spark.readStream.format("json").schema(my_schema).load("/in/path")
// Perform transformations and then write…
    sdf.writeStream.trigger(Trigger.Once).format("parquet").start("/out/path")

二，RunOnce相比Batch高效之處

1，Bookkeeping

當運行一個執行增量更新的批處理作業時，通常要處理哪些數據是更新的，哪些是該處理的，哪些是不該處理的。Structured Streaming已經為你做好了這一切，在處理一般流式應用程序時，你應該只關心業務邏輯，而不是低級的Bookkeeping。

2，表級原子性

大數據處理引擎，最重要的性質是它如何容忍失誤和失敗。ETL作業可能(實際上常會)失敗。如果，你的工作失敗了，那么你需要確保你的工作產出被清理干凈，否則在你的下一次成功的工作之后你會得到重復的或者垃圾的數據。使用Structured Streaming編寫基于文件的表時，Structured Streaming將每個作業創建的所有文件在每次成功的出發后提交到log中。當Spark重新讀取表時，會通過log來識別哪些文件是有效的。這樣可以確保因失敗引入的垃圾不會被下游的應用程序所消費。

3，夸runs的狀態操作

如果，你的數據流有可能產生重復的記錄，但是你要實現一次語義，如何在batch處理中來實現呢？通過Structured Streaming，可以使用dropDuplicates()來實現去重。配置watermark足夠長，包含若干Streaming job的runs，可以保證你不會夸runs處理到重復的數據。

4，成本節約

運行一個24*7的Streamingjob很浪費。可能有些情況，數據計算有些延遲是可以接受的，或者數據本身就會以每小時或者每天為周期產生。為了獲得Structured Streaming所有上述描述的好處，你可能會人為需要一直占用集群運行程序，但是現在，使用僅執行一次的Trigger，就可以不必要一直占用集群了。

看完上述內容，你們掌握Spark Structured Streaming的高效處理RunOnceTrigger是怎樣進行的的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！