title: Spark Kafka Direct DStream - How many executors and RDD partitions in yarn-cluster mode if num-executors is set?
body: <p><br>
  I am trying to use Spark Kafka Direct Stream approach. It simplifies parallelism by creating as many RDD partitions as kafka topic partition, as stated in this <a href="https://spark.apache.org/docs/latest/streaming-kafka-integration.html" rel="noreferrer">doc</a>. And base on my understanding, spark will create one executor for each RDD partition to do the computation. </p>
  
  <p>So when I submit the application in yarn-cluster mode, and specify option <em>num-executors</em> to a different value to the number of partitions, how many executors there will be?</p>
  
  <p>For example, there is a kafka topic with 2 partition, and I specify <em>num-executors</em> to 4:</p>
  
  <pre><code>export YARN_CONF_DIR=$HADOOP_HOME/client_conf
  
  ./bin/spark-submit \
  --class playground.MainClass \
  --master yarn-cluster \
  --num-executors 4 \
  ../spark_applications/uber-spark-streaming-0.0.1-SNAPSHOT.jar \
  127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095 topic_1
  </code></pre>
  
  <p>I do give it a try and find out the number of executors is 4, and each executor do reads and processes data from kafka. Why? There is only 2 partitions in kafka topic, How does 4 executors read from the kafka topic, which only has 2 partitions?</p>
  
  <p>Below is the details of the spark application and logs.</p>
  
  <p><strong>My spark application</strong>, which prints received messages ( in <em>flatMap</em> method)  from kafka in every executor:</p>
  
  <pre><code>    ...
      String brokers = args[0];
      HashSet&lt;String&gt; topicsSet = new HashSet&lt;String&gt;(Arrays.asList(args[1].split(",")));
      kafkaParams.put("metadata.broker.list", brokers);
  
      JavaPairInputDStream&lt;String, String&gt; messages =
          KafkaUtils.createDirectStream(jssc, String.class, String.class, StringDecoder.class, StringDecoder.class,
              kafkaParams, topicsSet);
  
      JavaPairDStream&lt;String, Integer&gt; wordCounts =
          messages.flatMap(new FlatMapFunction&lt;Tuple2&lt;String, String&gt;, String&gt;()
          {
              public Iterable&lt;String&gt; call(Tuple2&lt;String, String&gt; tuple) throws Exception
              {
                  System.out.println(String.format("[received from kafka] tuple_1 is %s, tuple_2 is %s", tuple._1(),
                      tuple._2())); // print the kafka message received  in executor
                  return Arrays.asList(SPACE.split(tuple._2()));
              }
  
          }).mapToPair(new PairFunction&lt;String, String, Integer&gt;()
          {
              public Tuple2&lt;String, Integer&gt; call(String word) throws Exception
              {
                  System.out.println(String.format("[word]: %s", word));
                  return new Tuple2&lt;String, Integer&gt;(word, 1);
              }
  
          }).reduceByKey(new Function2&lt;Integer, Integer, Integer&gt;()
          {
              public Integer call(Integer v1, Integer v2) throws Exception
              {
                  return v1 + v2;
              }
  
          });
  
      wordCounts.print();
  
      Runtime.getRuntime().addShutdownHook(new Thread(){
          @Override
          public void run(){
              System.out.println("gracefully shutdown Spark!");
              jssc.stop(true, true);
          }
      });
      jssc.start();
      jssc.awaitTermination();
  </code></pre>
  
  <p><strong>My Kafka topic</strong>, with 2 partitions. String <em>"hello hello word 1", "hello hello word 2", "hello hello word 3",</em> ... are sent to the topic.</p>
  
  <pre><code>Topic: topic_2  PartitionCount:2    ReplicationFactor:2 Configs:
  Topic: topic_2  Partition: 0    Leader: 3   Replicas: 3,1   Isr: 3,1
  Topic: topic_2  Partition: 1    Leader: 1   Replicas: 1,2   Isr: 1,2
  </code></pre>
  
  <p><strong>Webconsle</strong>:
  <img src="https://i.stack.imgur.com/S17a5.png" alt="enter image description here"></p>
  
  <p><strong>console output of executor 1</strong>:</p>
  
  <pre><code>...
  [received from kafka] tuple_1 is null, tuple_2 is hello hello world 12
  [word]: hello
  [word]: hello
  [word]: world
  [word]: 12
  ...
  </code></pre>
  
  <p><strong>console output of executor 2</strong>:</p>
  
  <pre><code>...
  [received from kafka] tuple_1 is null, tuple_2 is hello hello world 2
  [word]: hello
  [word]: hello
  [word]: world
  [word]: 2
  ...
  </code></pre>
  
  <p><strong>console output of executor 3</strong>:</p>
  
  <pre><code>...
  [received from kafka] tuple_1 is null, tuple_2 is hello hello world 3
  [word]: hello
  [word]: hello
  [word]: world
  [word]: 3
  ...
  </code></pre>