MongoDBObject tidak ditambahkan ke dalam rrd foreach loop casbah scala Apache spark

Perhitungan pada RDD didistribusikan ke cluster. Anda tidak dapat memperbarui variabel yang dibuat di luar penutupan operasi RDD dari dalam RDD. Mereka pada dasarnya berada di dua tempat yang berbeda:Variabel dibuat di driver Spark dan diakses di pekerja dan harus diperlakukan sebagai hanya-baca.

Spark mendukung cummulator terdistribusi yang dapat digunakan dalam kasus ini:Spark Cummulator

Opsi lain (yang saya lebih suka) adalah mengubah aliran RDD ke dalam format data yang diinginkan dan menggunakan foreachRDD metode untuk menyimpannya ke dalam penyimpanan sekunder. Ini akan menjadi cara yang lebih fungsional untuk mendekati masalah. Kira-kira akan terlihat seperti ini:

  val filteredStream = twitterStream.filter(entry => filters.exists(term => entry.getText.getStatus.contains(term)))
  val filteredStreamWithTs = filteredStream.map(x => ((DateTime.now.toString(), x)))
  filteredStreamWithTs.foreachRdd(rdd => // write to Mongo)