Spark Streamingを使用して計算結果を定期的に保存しますか？

Question

私は、入力データで計算されたいくつかの集計に基づいて、リアルタイムダッシュボードを表示するという要件に取り組んでいます。

私はSpark/Spark Streamingを探索し始めました。Spark Integrationをマイクロバッチでリアルタイムで計算し、UIダッシュボードに提供することができます。

私のクエリは、Spark Integrationジョブの開始後、いつでも停止またはクラッシュし、最後に処理していた位置から再開する方法です。私はSparkが社内の状態を維持していることを理解し、受け取ったすべての新しいデータについてその状態を更新します。しかし、それが再開されたとき、その状態は消えないでしょう。

Sparkが再び再起動したときにSparkが処理を再開できるように、実行中の合計/結果を定期的に保持する必要があると感じています。しかし、スパークストリーミングでどうすればいいのか分かりません。

しかし、Spark Streamingがデフォルトで使用し始めているので、データが失われないようにするかどうかはわかりません。

誰かが同様のシナリオに直面した場合は、これに対処する方法について考えてください。

Answer 1

スパークストリーミングはコンシューマアプリケーションとして機能します。リアルタイムでは、データを一部のデータストアに保存できるKafkaのトピックからデータを取得しています。これは、あなたがTwitterのストリームからデータを読み込んでいる場合にも当てはまります。オフセットを保存するために、そしてアプリケーションがクラッシュまたは再起動した場合、以下のポストに従うことができます。

http://aseigneurin.github.io/2016/05/07/spark-kafka-achieving-zero-data-loss.html

https://www.linkedin.com/pulse/achieving-exactly-once-semantics-kafka-application-ishan-kumar

Answer 2

キーポイント：

詳細

チェックポイントを有効レシーバ

のためにログ先行書き込み可能

1. はWALを有効にする：spark.streaming.receiver.writeAheadLog.enable true
2. を設定したチェックポイント

チェックポイントは定期的に信頼性の高いストレージにアプリの状態を記述することで可能になります。アプリケーションが失敗すると、チェックポイントファイルから回復することができます。 チェックポイントを書き込むには、このように書きます：

ssc.checkpoint("checkpoint.path") 

がチェックポイントから読み込むには：

def main(args: Array[String]): Unit = { 
    val ssc = StreamingContext.getOrCreate("checkpoint_path",() => createContext()) 

    ssc.start() 
    ssc.awaitTermination() 
} 

createContext機能で、あなたはSSCを作成し、独自のロジックを実行する必要があります。例：

def createContext(): StreamingContext = { 
    val conf = new SparkConf() 
    .setAppName("app.name") 
    .set("spark.streaming.stopGracefullyOnShutdown", "true") 

    val ssc = new StreamingContext(conf, Seconds("streaming.interval")) 
    ssc.checkpoint("checkpoint.path") 

    // your code here 

    ssc 
} 

ドライバ/エグゼキュータの障害からの復旧を含むスパークストリーミングアプリケーションの展開方法に関する必要な手順についてのドキュメントです。

https://spark.apache.org/docs/1.6.1/streaming-programming-guide.html#deploying-applications