def scioHashJoin()

in src/main/scala/com/spotify/bdrc/pipeline/JoinLogAndMetadata.scala [115:126]

• 12 lines of code
• 1 McCabe index (conditional complexity)

  def scioHashJoin(
    left: SCollection[LogEvent],
    right: SCollection[UserMeta]
  ): SCollection[(String, Double)] = {
    import com.twitter.algebird.AveragedValue
    val lhs = left.map(e => (e.user, e.track))
    val rhs = right.map(u => (u.user, u.age.toDouble))
    lhs
      .hashJoin(rhs)
      .values
      .aggregateByKey(AveragedValue.aggregator)
  }