Combine.perKey lifting test

SlowFanOutThenCombine.scala

import com.spotify.scio._
import org.slf4j.Logger
import org.slf4j.LoggerFactory

object SlowFanOutThenCombine {
  val LOG: Logger = LoggerFactory.getLogger(this.getClass)

  def slowFanOut(n: Int): Iterator[(Int, (Int, Int))] = new Iterator[(Int, (Int, Int))] {
    var i: Int = 0

    override def hasNext: Boolean = i < 40

    override def next(): (Int, (Int, Int)) = {
      i += 1
      Thread.sleep(1000)
      LOG.info(s"Emit ($n, $i)")
      (n, (n, i))
    }
  }

  def seqOp(acc: (Int, Int), x: (Int, Int)): (Int, Int) = {
    LOG.info(s"Seq $acc, $x")
    (acc._1.max(x._1), acc._2 + x._2)
  }

  def combOp(x: (Int, Int), y: (Int, Int)): (Int, Int) = {
    LOG.info(s"Comb $x, $y")
    (x._1.max(y._1), x._2 + y._2)
  }

  def main(cmdlineArgs: Array[String]): Unit = {
    val (sc, _) = ContextAndArgs(cmdlineArgs)

    val f =
      sc.parallelize(1 to 8)
      .flatMap(slowFanOut)
      .aggregateByKey((0, 0))(seqOp, combOp)
      .materialize

    sc.close()
    for (t <- f.waitForResult().value) {
      println(t)
    }
  }
}