0
A
回答
0
在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()
+0
我不認爲這回答我的問題。我的問題是關於如何計算特定領域的變化。 –
相關問題
- 1. 檢測流數據中的更改
- 2. 使用Spark和R進行時間序列趨勢檢測
- 3. K-Means使用Apache Spark的時間序列數據
- 4. 刪除冗餘列在Spark數據幀時間序列數據
- 5. 如何檢測空間流的數量
- 6. 如何在使用JSON.NET進行序列化時更改數據?
- 7. 如何使用jquery檢測多列表框中的更改
- 8. 如何檢測Android應用程序的無用數據流量?
- 9. 如何檢測iCloud中的文件更改時間?
- 10. 數據庫更改檢測
- 11. 檢測postgresql中的數據更改
- 12. 使用Oracle數據庫更改通知檢測更改
- 13. 如何使用jQuery檢測URL更改
- 14. 當您更改表格時,如何檢測列的數據類型?
- 15. 如何使用Spark Streaming從序列文件中讀取數據
- 16. 檢測時間序列數據的變化
- 17. 如何在mongodb中有效存儲快速更改的時間序列數據?
- 18. 如何測量Spark GraphX應用程序的執行時間?
- 19. 如何在調試時檢測實例數據屬性更改?
- 20. 如何檢測Control.PreferredSize何時更改?
- 21. 如何檢測服務何時更改?
- 22. 如何檢測輸入何時更改?
- 23. Javascript。如何檢測window.document何時更改?
- 24. 使用Node.js檢測Mongodb中的數據更改
- 25. 使用動態綁定數據檢測textarea中的更改?
- 26. 如何檢測時間序列數據的變化是否不再顯着?
- 27. 如何在時序圖中啓用時間序列數據?
- 28. 使用時間序列數據預測事件的發生
- 29. 預測時間序列數據
- 30. PHP DRUPAL:如何檢測數據庫中表格的更改
考查狀態轉換,如'mapWithState'其可以彙總microbatches結果。 –
當然,會去看看。謝謝, –