title: Flink: Syncronize/connect two streams
body: <p>I have two streams, the first, with CEP library, it´s used it to detect patterns ON-OFF. In the second stream are values that I have to add. I'm looking for way to connect the two streams, so that when I receive an OFF frame, I return the device status change, with the ultimate value of the sum of the second type of frames: </p>
  
  <pre><code>Device 311 State Change:Ignition OFF  Value:35.91
  </code></pre>
  
  <p>My code:</p>
  
  <pre><code>// IGNITION pattern
  val ig = timedValue.filter(_._2 == "Ignition").keyBy(_._4)
  val patternIG_ON: Pattern[(Long,String,String,Long), _] = Pattern.begin[(Long,String,String,Long)]("start").where(_._3 == "OFF").next("end").where(_._3 == "ON")
  val patternStream: PatternStream[(Long,String,String,Long)] = CEP.pattern(ig, patternIG_ON)
  val patternIG_OFF: Pattern[(Long,String,String,Long), _] = Pattern.begin[(Long,String,String,Long)]("start").where(_._3 == "ON").next("end").where(_._3 == "OFF")
  val patternStreamIG_OFF: PatternStream[(Long,String,String,Long)] = CEP.pattern(timedValue, patternIG_OFF)
  
  def selectFn(pattern : mutable.Map[String,(Long,String,String,Long)]): String = {
    val startEvent = pattern.get("start").get
    val endEvent = pattern.get("end").get
    "Device "+startEvent._4 +" State Change: "+startEvent._2+" ON"
  }
  def selectFnOFF(pattern : mutable.Map[String,(Long,String,String,Long)]): String = {
    val startEvent = pattern.get("start").get
    val endEvent = pattern.get("end").get
    "Device "+startEvent._4 +" State Change: "+startEvent._2+" OFF "
  }
  
  val patternStreamSelected = patternStream.select(selectFn(_)).print()
  val suma: org.apache.flink.streaming.api.scala.DataStream[(Long, Double)] = odo.map(s =&gt; (s._4,s._3)).keyBy(_._1).mapWithState(
    (in, state: Option[(Int, Double)]) =&gt; state match {
      case Some((count, sum)) =&gt; ((in._1,(sum + in._2).toDouble), Some((count + 1, sum + in._2)))
      case None =&gt; ((in._1, in._2), Some(1, in._2))
    })
  val patternStreamSelectedOFF = patternStreamIG_OFF.select(selectFnOFF(_)).connect(suma).keyBy(_._1, _._1).map(l =&gt; l + " Value:", r =&gt; r._2).print()
  </code></pre>
  
  <p><strong>Update</strong>: My Result:</p>
  
  <pre><code>24.2
  23.5
  44.010000000000005
  67.05000000000001
  Device 311 State Change3:Ignition ON
  (311,Device 311 State Change: Ignition OFF ) Value:
  </code></pre>
  
  <p>the first number (24.2) is from another device, it should not appear...</p>
  
  <p>Any idea what is the best approach??
  Thank you</p>