スパークストリーミングのさまざまなマイクロバッチからの集計データ

Question

私は毎分スパークストリーミング（Kafkaからの読書）を使用していくつかのメトリックを集約して見つけようとしています。その分のデータを集計することができます。どのように私は今日のバケツを持つことができ、その日のすべての分のすべての集計値を合計することができますか？

私はデータフレームを持っており、私はこれに似た何かをやっています。

sampleDF = spark.sql("select userId,sum(likes) as total from likes_dataset group by userId order by userId") 

Answer 1

あなたは

サンプルコード

import spark.implicits._ 

val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String } 

    val windowedCounts = words 
     .withWatermark("timestamp", "10 minutes") 
     .groupBy(
      window($"timestamp", "10 minutes", "5 minutes"), 
      $"word") 
     .count() 

Answer 2

構造化ストリーミングプログラミングから「Watermarking」機能を利用することができ、私は何が起こっている考え出しました。 Sparkのステートフルなストリーミングについて知る必要があり、それが私を助けました。

私がしなければならなかったすべては、私はマイクロバッチの新しい集計データと古い集計データをマージする方法を言うために、このupdateTotalCount関数を書かなければならなかった

running_counts = countStream.updateStateByKey(updateTotalCount, initialRDD=initialStateRDD) 

です。私の場合、更新機能は次のようになります。

def updateTotalCount(currentCount, countState): 
    if countState is None: 
     countState = 0 
    return sum(currentCount) + countState