title: Spark streaming join Kafka topics comparison
body: <p>We needed to implement join on Kafka topics with consideration of late data  or "not in join", meaning data  that come late on the stream or not  in join will not be dropped/lost but will be marked as a timeout,</p>
  
  <p>the result of the join is produced to output Kafka topic ( with a timeout filed if occurred).</p>
  
  <p>(spark 2.1.1 in standalone deployment, Kafka 10 )</p>
  
  <p>Kafka in topics: X, Y,...  out  topics result will look like:</p>
  
  <pre><code>{
      "keyJoinFiled": 123456,
      "xTopicData": {},
      "yTopicData": {},
      "isTimeOutFlag": true
  }
  </code></pre>
  
  <p>I found three solutions wrote them here, 1 and 2 from spark streaming official documentation but are not relevant to us ( data not in join Dtsream, arrive "business time" late, is dropped/lost) but I wrote them for comparison.</p>
  
  <p>From what we saw there are not too many examples for Kafka join topic with a stateful operation add some code here for review:</p>
  
  <p>1) According to spark streaming documentation,</p>
  
  <pre><code>https://spark.apache.org/docs/2.1.1/streaming-programming-guide.html:   
   val stream1: DStream[String, String] = 
   val stream2: DStream[String, String] = 
   val joinedStream = stream1.join(stream2)
  </code></pre>
  
  <p>This will join data from both stream batches duration, but data arrive "business time" late/not in join will be dropped/lost.</p>
  
  <p>2) Window join:</p>
  
  <pre><code>val leftWindowDF = kafkaStreamLeft.window(Minutes(input_parameter_time))
  val rightWindowDF = kafkaStreamRight.window(Minutes(input_parameter_time))
  leftWindowDF.join(rightWindowDF).foreachRDD...
  </code></pre>
  
  <p>2.1) In our case we need to use Tumbling window in consideration for 
         spark streaming batch interval.
    2.2) Need to save a lot of data in Memory/Disk, for example, 30-60 min 
         window
    2.3) And again data arrive late/not in the window/not in the join is 
         dropped/lost.
         * Since spark 2.3.1 Structured streaming stream to stream join is 
           supported, but we encounter a bug with not cleaning HDFS state 
           store, as a result, a job was falling every few hours on OOM, 
           resolved in 2.4
          ,<a href="https://issues.apache.org/jira/browse/SPARK-23682" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/SPARK-23682</a> 
          (use of Rocksdb ,or CustomStateStoreProvider HDFS state store).  </p>
  
  <p>3) Using stateful operation mapWithState for join Kafka topics Dstreams 
     with tumbling window and 30 min timeout for late data,
     all data produced to output topics contains joined messages from all 
     topics if join occurred or part of topic data if no
     join occurred in 30 min ( mark with is_time_out flag) </p>
  
  <p>3.1) Creating 1..n Dstream per topic, convert to Key value/Unioned 
       records with join filed as a key and tumbling window.
       creating a catch-all scheme.
   3.2)Union all streams 
   3.3)Run on union stream mapWithState with function - will actually do the 
       join/ mark timeout.</p>
  
  <p>Great example for stateful join from databricks (spark 2.2.0): 
  <a href="https://www.youtube.com/watch?time_continue=1858&amp;v=JAb4FIheP28" rel="nofollow noreferrer">https://www.youtube.com/watch?time_continue=1858&amp;v=JAb4FIheP28</a></p>
  
  <p>Adding a sample code that is running/testing.</p>
  
  <pre><code> val kafkaParams = Map[String, Object](
      "bootstrap.servers" -&gt; brokers,
      "key.deserializer" -&gt; classOf[StringDeserializer],
      "value.deserializer" -&gt; classOf[StringDeserializer],
      "group.id" -&gt; groupId,
      "session.timeout.ms" -&gt; "30000"
    )
  
    //Kafka xTopic DStream
    val kafkaStreamLeft = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](leftTopic.split(",").toSet, kafkaParams)
    ).map(record =&gt; {
      val msg:xTopic = gson.fromJson(record.value(),classOf[xTopic])
      Unioned(Some(msg),None,if (msg.sessionId!= null) msg.sessionId.toString else "")
    }).window(Minutes(leftWindow),Minutes(leftWindow))
  
    //Kafka yTopic DStream
    val kafkaStreamRight = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](rightTopic.split(",").toSet, kafkaParams)
    ).map(record =&gt; {
      val msg:yTopic = gson.fromJson(record.value(),classOf[yTopic])
      Unioned(None,Some(msg),if (msg.sessionId!= null) msg.sessionId.toString else "")
    }).window(Minutes(rightWindow),Minutes(rightWindow))
  
    //convert stream to key, value pair and filter empty session id.
    val unionStream = kafkaStreamLeft.union(kafkaStreamRight).map(record =&gt;(record.sessionId,record))
      .filter(record =&gt; !record._1.toString.isEmpty)
    val stateSpec = StateSpec.function(stateUpdateF).timeout(Minutes(timeout.toInt))
  
    unionStream.mapWithState(stateSpec).foreachRDD(rdd =&gt; {
      try{
        if(!rdd.isEmpty()) rdd.foreachPartition(partition =&gt;{
          val props = new util.HashMap[String, Object]()
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  
          val producer = new KafkaProducer[String, String](props)
          //send to kafka result JSON.
          partition.foreach(record =&gt; {
            if(record!=null &amp;&amp; !"".equals(record) &amp;&amp; !"()".equals(record.toString) &amp;&amp; !"None".equals(record.toString) ){
              producer.send(new ProducerRecord[String, String](outTopic, null, gson.toJson(record)))
            }
          })
          producer.close()
        })
      }catch {
        case e: Exception  =&gt; {
          logger.error(s""""error join topics :${leftTopic} ${rightTopic} to out topic ${outTopic}""")
          logger.info(e.printStackTrace())
        }
      }})
  
  //mapWithState function that will be called on each key occurrence with new items in newItemValues and state items if exits.
  
  def stateUpdateF = (keySessionId:String,newItemValues:Option[Unioned],state:State[Unioned])=&gt; {
      val currentState = state.getOption().getOrElse(Unioned(None,None,keySessionId))
  
      val newVal:Unioned = newItemValues match {
        case Some(newItemValue) =&gt; {
          if (newItemValue.yTopic.isDefined)
            Unioned(if(newItemValue.xTopic.isDefined) newItemValue.xTopic else currentState.xTopic,newItemValue.yTopic,keySessionId)
          else if (newItemValue.xTopic.isDefined)
            Unioned(newItemValue.xTopic, if(currentState.yTopic.isDefined)currentState.yTopic else newItemValue.yTopic,keySessionId)
          else newItemValue
        }
        case _ =&gt; currentState //if None = timeout =&gt; currentState
      }
  
      val processTs = LocalDateTime.now()
      val processDate = dtf.format(processTs)
      if(newVal.xTopic.isDefined &amp;&amp; newVal.yTopic.isDefined){//if we have a join remove from state
        state.remove()
        JoinState(newVal.sessionId,newVal.xTopic,newVal.yTopic,false,processTs.toInstant(ZoneOffset.UTC).toEpochMilli,processDate)
      }else if(state.isTimingOut()){//time out do no try to remove state manually ,it's removed automatically.
          JoinState(newVal.sessionId, newVal.xTopic, newVal.yTopic,true,processTs.toInstant(ZoneOffset.UTC).toEpochMilli,processDate)
      }else{
        state.update(newVal)
      }
    }
  
    //case class for kafka topics data.(x,y topics ) join will be on session id filed.
    case class xTopic(sessionId:String,param1:String,param2:String,sessionCreationDate:String)
    case class yTopic(sessionId:Long,clientTimestamp:String)
    //catch all schema : object that contains both kafka input fileds topics and key valiue for join.
    case class Unioned(xTopic:Option[xTopic],yTopic:Option[yTopic],sessionId:String)
    //class for  output result of join stateful function.
    case class JoinState(sessionId:String, xTopic:Option[xTopic],yTopic:Option[yTopic],isTimeOut:Boolean,processTs:Long,processDate:String)
  </code></pre>
  
  <p>I will be happy for some review.
  sorry for the long post. </p>