如何使用Spark流檢測時間序列數據中的更改

Question

我在卡夫卡有連續的數據流。我想要統計數據流中的列值已更改的次數。

我應該使用哪種算法？

Answer 1

在Spark 2.0中使用Structured Streaming，處理流式DataFrame非常接近正常的DataFrame。在以下測試示例中，添加新批次數據時，值計數將打印到控制檯。

val wordCounts = words.groupBy("value").count() 
val query = wordCounts.writeStream 
    .outputMode("complete") 
    .format("console") 
    .start() 

我們還可以創建自己的StreamSinkProvider來決定在有新的批處理數據出現時該怎麼做。

class CustomSinkProvider extends StreamSinkProvider { 
    def createSink(
        sqlContext: SQLContext, 
        parameters: Map[String, String], 
        partitionColumns: Seq[String], 
        outputMode: OutputMode): Sink = { 
    new Sink { 
     override def addBatch(batchId: Long, data: DataFrame): Unit = { 
     // Do something. 
     } 
    } 
    } 
} 

然後使用以下代碼來使用CustomSinkProvider

val query = wordCounts.writeStream 
    .outputMode("complete") 
    .format(classOf[CustomSinkProvider].getCanonicalName) 
    .start()