title: Spark Streaming code to read from Kafka broker and calculate average of numbers by mapWithState
body: <p>I'm receiving my value from Kafka in the form of <code>character,number</code> in a random manner which is generated by another program. An example of values I receive:</p>
  <pre><code>a,4
  b,3
  d,7
  f,5
  b,2
  ...
  </code></pre>
  <p>Here is the program that generates these values and sends it over Kafka topic:</p>
  <pre><code>package generator
  
  import java.util.{Date, Properties}
  import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, ProducerConfig}
  import scala.util.Random
  import kafka.producer.KeyedMessage
  
  object ScalaProducerExample extends App {
      def getRandomVal: String = {
          val i = Random.nextInt(alphabet.size)
          val key = alphabet(i)
          val value = Random.nextInt(alphabet.size)
          key + &quot;,&quot; + value
      }
  
      val alphabet = 'a' to 'z'
      val events = 10000
      val topic = &quot;avg&quot;
      val brokers = &quot;localhost:9092&quot;
      val rnd = new Random()
  
      val props = new Properties()
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
      props.put(ProducerConfig.CLIENT_ID_CONFIG, &quot;ScalaProducerExample&quot;)
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, &quot;org.apache.kafka.common.serialization.StringSerializer&quot;)
      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, &quot;org.apache.kafka.common.serialization.StringSerializer&quot;)
      val producer = new KafkaProducer[String, String](props)
  
      while (true) {
          val data = new ProducerRecord[String, String](topic, null, getRandomVal)
          producer.send(data)
          print(data + &quot;\n&quot;)
      }
  
      producer.close()
  }
  </code></pre>
  <p>My task is to show the up-to-date average for every character based on the sum and number of received values for it from the beginning until now.
  I wrote this code for this task and I am receiving from Kafka successfully:</p>
  <pre><code>package DirectKafkaWordCount
  
  import org.apache.kafka.clients.consumer.ConsumerConfig
  import org.apache.kafka.common.serialization.StringDeserializer
  
  import org.apache.spark.SparkConf
  import org.apache.spark.streaming._
  import org.apache.spark.streaming.kafka010._
  
  case class Data(key: String, count: Int)
  
  object DirectKafkaWordCount {
    def main(args: Array[String]): Unit = {
      val Array(brokers, topics) = args
      val sparkConf = new SparkConf().setMaster(&quot;local[4]&quot;).setAppName(&quot;DirectKafkaWordCount&quot;)
      val ssc = new StreamingContext(sparkConf, Seconds(2))
      ssc.checkpoint(&quot;_checkpoint&quot;)
      val topicsSet = topics.split(&quot;,&quot;).toSet
      val kafkaParams = Map[String, Object](
        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -&gt; brokers,
        ConsumerConfig.GROUP_ID_CONFIG -&gt; &quot;1&quot;,
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -&gt; classOf[StringDeserializer],
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -&gt; classOf[StringDeserializer])
      val messages = KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams))
  
      val pairs = messages.map(_.value).map(x =&gt; (x.split(&quot;,&quot;)(0), x.split(&quot;,&quot;)(1).toInt))
      val wc = pairs.mapWithState(StateSpec.function((key: String, value: Option[Int], state: State[String]) =&gt; {
        val newNum = value.getOrElse(0)
        val sData = state.getOption.getOrElse(&quot;a,0,0&quot;)
        var count = sData.split(&quot;,&quot;)(1).toInt
        var sum = sData.split(&quot;,&quot;)(2).toInt
        sum = sum + newNum
        count = count + 1
        val output = key + &quot;,&quot; + count.toString + &quot;,&quot; + sum.toString
        state.update(output)
        output
      }))
      wc.map(process _).print()
      ssc.start()
      ssc.awaitTermination()
    }
  
    def process(s: String): String = {
      var count = s.split(&quot;,&quot;)(1).toInt
      var sum = s.split(&quot;,&quot;)(2).toInt
      s.split(&quot;,&quot;)(0) + &quot;,&quot; + (sum / count).toString
    }
  }
  </code></pre>
  <p>My problem is that the average for every character becomes the constant number 12. Is there sth wrong with the mapWithState function? How can I fix it? Something that makes me suspicious is that there isn't only one entry per character in the output, there may be 3 or 4 entries per character. A sample output:</p>
  <pre><code>-------------------------------------------
  Time: 1651560488000 ms
  -------------------------------------------
  d,12
  t,12
  h,12
  t,12
  h,12
  x,12
  d,12
  h,12
  p,12
  p,12
  ...
  </code></pre>