title: How to validate every row of streaming batch?
body: <p>Need to validate each row of Streaming Dataframe (consumed through readStream(kafka) - Getting error </p>
  
  <pre>Queries with streaming sources must be executed with writeStream.start()</pre>
  
  <p>as it is not allowing to validate row by row</p>
  
  <p>I have created spark batch job to consume data from Kafka , validated each row against HBase data, another set of validations based on rowkey and created a dataframe out of it. But here I need to handle the Kafka offset manually in the code. </p>
  
  <p>To avoid the offset handling, am trying to use spark structural Streaming but there am not able to validate row by row.</p>
  
  <p>writestream.foreach (foreachwriter) is good to sink with any external datasource or writing to kafka. </p>
  
  <p>But in my case, I need to validate each row and create a new dataframe based on my validation. foreachwriter - process is not allowing to collect the data using other external classes/list. </p>
  
  <p>Errors:</p>
  
  <p>Getting this error when I tried to access the streaming dataframe to validate</p>
  
  <pre>Queries with streaming sources must be executed with writeStream.start();</pre>
  
  <p><pre> Task is not serializable </pre> when I tried to create a list out of foreach(foreachwriter extended object). Will update with more details (as I am trying other options)</p>
  
  <p>I am trying to achieve spark batch using <pre> writerstream.trigger(Trigger.once) </pre> with checkpointlocation</p>
  
  <p>Updating with Spark batch and Structural Streaming Code.</p>
  
  <pre><code>      .read
        .format("kafka")
        .option("kafka.bootstrap.servers", kafkaBootStrap)
        .option("subscribePattern", kafkaSubTopic)                
        .option("startingOffsets", "earliest") 
        //.option("endingOffsets", "latest")
        .load()
  
     rawData.collect.foreach(row =&gt; {
        if (dValidate.dValidate(row)) {
          validatedCandidates += (row.getString(0))
        }       
  
  
  ==================== in the above code I need to handle the offset manually for rerun so decided to use structural streaming.============
  
  
     val rawData = spark.readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", kafkaBootStrap)
          .option("subscribe", kafkaSubTopic)
          .option("enable.auto.commit", "true")
          .option("startingOffsets","latest")
          .option("minPartitions", "10")  
          .option("failOnDataLoss", "true")          
          .load()   
  
  
     val sInput = new SinkInput(validatedCandidates,dValidate)
  
    rawData.writeStream
       .foreach(sInput)
       .outputMode(OutputMode.Append())
       .option("truncate", "false")
       .trigger(Trigger.Once())
       .start()
  </code></pre>
  
  <p>am getting "Task not serialized" error in here.
  with class SinkInput , I am trying to do the same collect operation with external dValidate instance        </p>