MongoDBO-objekt läggs inte till inuti en rrd foreach loop casbah scala apache spark

Beräkningarna på RDD:er är fördelade över klustret. Du kan inte uppdatera en variabel som skapades utanför RDD-operationens stängning inifrån RDD. De finns i princip på två olika platser:Variabeln skapas i Spark-drivrutinen och nås i arbetarna och ska behandlas som skrivskyddad.

Spark stöder distribuerade cummulatorer som kan användas i det här fallet:Spark Cummulators

Ett annat alternativ (det jag föredrar) är att omvandla RDD-strömmen till önskat dataformat och använda foreachRDD metod för att bevara den i sekundär lagring. Detta skulle vara ett mer funktionellt sätt att närma sig problemet. Det skulle ungefär se ut så här:

  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)