title: Mixing Spark Structured Streaming API and DStream to write to Kafka
body: <p>I've recently noticed I have a confusion regarding Spark Streaming (I'm currently learning Spark).</p>
  <p>I am reading data from a Kafka topic like this:</p>
  <pre><code>val stream = KafkaUtils.createDirectStream[String, String](
        streamingContext,
        PreferConsistent,
        Subscribe[String, String](topics, kafkaParams)
      )
  </code></pre>
  <p>Which creates a DStream.</p>
  <p>In order to work with event-time (and not processing-time) I did this:</p>
  <pre><code>outputStream
        .foreachRDD(rdd =&gt; {
            rdd.toDF().withWatermark(&quot;timestamp&quot;, &quot;60 seconds&quot;)
              .groupBy(
                window($&quot;timestamp&quot;, &quot;60 seconds&quot;, &quot;10 seconds&quot;)
              )
              .sum(&quot;meterIncrement&quot;)
              .toJSON
              .toDF(&quot;value&quot;)
              .writeStream
              .format(&quot;kafka&quot;)
              .option(&quot;kafka.bootstrap.servers&quot;, &quot;localhost:9092&quot;)
              .option(&quot;topic&quot;, &quot;taxi-dollar-accurate&quot;)
              .start()
          )
        })
  </code></pre>
  <p>And I get the error</p>
  <blockquote>
  <p>'writeStream' can be called only on streaming Dataset/DataFrame</p>
  </blockquote>
  <p>Which surprised me, because the source of the DF is a DStream. Anyway, I managed to solve this by changing .writeStream to .write and .start() to .save().</p>
  <p>But I got the feeling that I lost the streaming power on that foreach somehow. Clearly that's why I am writing this question. Is this a correct approach? I've seen other scripts that use</p>
  <pre><code>val df = spark
    .readStream
    .format(&quot;kafka&quot;)
    .option(&quot;kafka.bootstrap.servers&quot;, &quot;host1:port1,host2:port2&quot;)
    .option(&quot;subscribe&quot;, &quot;topic1&quot;)
    .load()
  df.selectExpr(&quot;CAST(key AS STRING)&quot;, &quot;CAST(value AS STRING)&quot;)
    .as[(String, String)]
  </code></pre>
  <p>But I don't know how different is this from just calling foreach on the DStream and then transforming each RDD to DF.</p>