spark_join.scala

def reducer(
  a: (List[BloodPressureEvent], List[HeartRateEvent]),
  b: (List[BloodPressureEvent], List[HeartRateEvent])
  ) = (a, b) match {
  case ((bp1, hr1), (bp2, h2)) => (bp1 ++ bp2, hr1 ++ h2)
}

val combinedStream =
  bloodPressureStream
    .map((event) => (event.userId, List(event)))
    .fullOuterJoin(heartRateStream.map((event) => (event.userId, List(event))))
    .map({
      case (id, (Some(x), Some(y))) => (id, (x, y))
      case (id, (_, Some(y))) => (id, (List(), y))
      case (id, (Some(x), _)) => (id, (x, List()))
      case (id, _) => (id, (List(), List()))
    })
    .reduceByKeyAndWindow(reducer(_, _), Minutes(60), Minutes(1))