title: Apache Spark with Apache Kafka integration
body: <p>I'm new to Spark, and I am using Spark Structured Streaming (Spark 3.0.0, scala 2.12) to read data from kafka topic</p>
  <p>I have 3 spark workers and the topic is divided into 3 partitions</p>
  <p>correct me if I'm wrong but as I read from different answers and from Spark documentation , spark use consumer groups to read from kafka , and should map the executors as 1:1 with partitions if I have the same number of each</p>
  <p>but when I check the consumer groups, I only find that one IP is consuming and it's the driver IP, so the data is not consumed directly from executors, and I found the tasks are only being divided for 2 executors and the third one is almost idle</p>
  <p>So I have two questions:</p>
  <ul>
  <li>why only the driver node is consuming from kafka? and Can I get the executors in each worker to directly consume data from kafka?</li>
  <li>Can I divide the workload over all the executors?</li>
  </ul>
  <pre><code>object DataProcessing extends Serializable with Logging {
  
    val topics =  &quot;&quot;
    val brokers = &quot;&quot;
  
  //window config =&gt; I'm using this during processing
    val windowDuration = s&quot;2 hours&quot;
    val slideDuration = s&quot;15 minutes&quot;
    
    @transient var streamingQuery: StreamingQuery = null
  
    val spark =
      SparkSession.builder
        .appName(&quot;app-name&quot;)
        .getOrCreate()
  
    import spark.implicits._
  
    def main(args: Array[String]): Unit = {
  
      val df = spark.readStream
        .format(&quot;kafka&quot;)
        .option(&quot;kafka.bootstrap.servers&quot;, brokers)
        .option(&quot;subscribe&quot;, topics)
        .option(&quot;startingOffsets&quot;, &quot;latest&quot;)
        .option(&quot;failOnDataLoss&quot;, &quot;false&quot;)
        .option(&quot;maxOffsetsPerTrigger&quot;, &quot;10000&quot;)
        .load()
      logDebug(&quot;Kafka Topic Data Schema =&gt; &quot;)
      df.printSchema()
  
   
      val ds = df
        .selectExpr(&quot;CAST(key AS STRING)&quot;, &quot;CAST(value AS STRING)&quot;)
        .as[(String, String)]
  
  //extracting columns =&gt; transformations and groupings =&gt; windowedData
  
      val writer = new AuroraSink(url, username, password)
      streamingQuery = windowedData.writeStream
        .queryName(&quot;data-processor&quot;)
        .foreach(writer)
        .outputMode(&quot;update&quot;)
        .option(&quot;checkpointLocation&quot;, s&quot;/data/$checkpoint&quot;)
        .trigger(ProcessingTime(&quot;10 seconds&quot;))
        .start()
  </code></pre>