title: How to save data to process later after stopping DirectStream in SparkStreaming?
body: <p>I am creating below KafkaDirectStream.</p>
  
  <pre><code>  val messages = KafkaUtils.createDirectStream[String, String](
           ssc,
           LocationStrategies.PreferConsistent,
           ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams))
  </code></pre>
  
  <p>Then saving the values as :</p>
  
  <pre><code>  val lines = messages.map(_.value)
  </code></pre>
  
  <p>Then stoping the streaming context when I have no further offset to consume as follows:</p>
  
  <pre><code>  lines.foreachRDD(rdd =&gt; {
        if(rdd.isEmpty()) {
          messages.stop()
          ssc.stop(false)
        } else {
  
        }
    })
  </code></pre>
  
  <p>Then I am printing the <code>lines</code> as follows:</p>
  
  <pre><code>    lines.print()
  </code></pre>
  
  <p>Then I am starting stream as: </p>
  
  <pre><code>    ssc.start()
  </code></pre>
  
  <p>It is working fine. It reads rdds and prints top 10 and stops messages stream and stop streaming context. But then when I execute the same line <code>lines.print()</code> it throws an exception saying cannot do new inputs, transform, or outputs after stoping streamingContext. </p>
  
  <p>How do I achieve my goal? I am running it in a spark-shell not as a binary (mandatory requirement).</p>
  
  <p>Here is what I actually want to achieve:</p>
  
  <p>1) Consume all json records from the kafka topic.</p>
  
  <p>2) Stop getting further records (It is guarenteed that after consuming, there won't be no new records added to Kafka topic, so don't want to keep proessing no records.)</p>
  
  <p>3) Do some preprocessing by extracting some fields from the JSON fields.</p>
  
  <p>4) Do further operation on the preprocessed data.</p>
  
  <p>5) Done.</p>