title: Using Spark Context in map of Spark Streaming Context to retrieve documents after Kafka Event
body: <p>I'm new to Spark.
  What I'm trying to do is retrieving all related documents from a Couchbase View with a given Id from Spark Kafka Streaming.</p>
  
  <p>When I try to get this documents form the Spark Context, I always have the error <code>Task not serializable</code>.</p>
  
  <p>From there, I do understand that I can't use nesting RDD neither multiple Spark Context in the same JVM, but want to find a work around.</p>
  
  <p>Here is my current approach:</p>
  
  <pre><code>package xxx.xxx.xxx
  
  import com.couchbase.client.java.document.JsonDocument
  import com.couchbase.client.java.document.json.JsonObject
  import com.couchbase.client.java.view.ViewQuery
  import com.couchbase.spark._
  
  import org.apache.spark.broadcast.Broadcast
  import _root_.kafka.serializer.StringDecoder
  import org.apache.kafka.clients.producer.{ProducerRecord, KafkaProducer}
  import org.apache.spark.{SparkConf, SparkContext}
  import org.apache.spark.streaming._
  import org.apache.spark.streaming.kafka._
  
  object Streaming {
    // Method to create a Json document from Key and Value
    def CreateJsonDocument(s: (String, String)): JsonDocument = {
      //println("- Parsing document")
      //println(s._1)
      //println(s._2)
      val return_doc = JsonDocument.create(s._1, JsonObject.fromJson(s._2))
      (return_doc)
      //(return_doc.content().getString("click"), return_doc)
    }
  
    def main(args: Array[String]): Unit = {
      // get arguments as key value
      val arguments = args.grouped(2).collect { case Array(k,v) =&gt; k.replaceAll("--", "") -&gt; v }.toMap
  
      println("----------------------------")
      println("Arguments passed to class")
      println("----------------------------")
      println("- Arguments")
      println(arguments)
      println("----------------------------")
  
      // If the length of the passed arguments is less than 4
      if (arguments.get("brokers") == null || arguments.get("topics") == null) {
        // Provide system error
        System.err.println("Usage: --brokers &lt;broker1:9092&gt; --topics &lt;topic1,topic2,topic3&gt;")
      }
  
      // Create the Spark configuration with app name
      val conf = new SparkConf().setAppName("Streaming")
      // Create the Spark context
      val sc = new SparkContext(conf)
      // Create the Spark Streaming Context
      val ssc = new StreamingContext(sc, Seconds(2))
  
      // Setup the broker list
      val kafkaParams = Map("metadata.broker.list" -&gt; arguments.getOrElse("brokers", ""))
      // Setup the topic list
      val topics = arguments.getOrElse("topics", "").split(",").toSet
      // Get the message stream from kafka
      val docs = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
  
      docs
        // Separate the key and the content
        .map({ case (key, value) =&gt; (key, value) })
        // Parse the content to transform in JSON Document
        .map(s =&gt; CreateJsonDocument(s))
        // Call the view to all related Review Application Documents
        //.map(messagedDoc =&gt; RetrieveAllReviewApplicationDocs(messagedDoc, sc))
        .map(doc =&gt; {
  
          sc.couchbaseView(ViewQuery.from("my-design-document", "stats").key(messagedDoc.content.getString("id"))).collect()
        })
        .foreachRDD(
            rdd =&gt; {
               //Create a report of my documents and store it in Couchbase
               rdd.foreach( println )
            }
          )
  
      // Start the streaming context
      ssc.start()
      // Wait for termination and catch error if there is a problem in the process
      ssc.awaitTermination()
    }
  }
  </code></pre>