如何將Spark Dstream元素添加到ArrayBuffer中

Question

我想將dstream元素添加到ArrayBuffer以進一步處理。但它不起作用。

var newBatchMeter = new ArrayBuffer[String]() 

newBatchReadingDstream.foreachRDD(rdd => { 
    for(item <- rdd.collect()) { 
    newBatchMeter += item.ID 
    } 
}) 

newBatchMeter.foreach(println) 

Answer 1

這不起作用的原因是foreach子句在worker中執行，而你的數組緩衝區在驅動程序中聲明。 收集DStream將無法​​正常工作，因爲它是一個無限流，您可以將數據保存到文件並稍後將其加載到ArrayBuffer中。

我會的東西，最好的辦法是找到一個方式做你想要使用DStream元

像與ArrayBuffer做處理：

newBatchReadingDstream.map(item => item.ID)