Reduce and ReduceByKey in BEAM 2.3.0

reduce.scala

import org.apache.beam.sdk.Pipeline
import org.apache.beam.sdk.transforms.Combine.BinaryCombineFn
import org.apache.beam.sdk.transforms.{Combine, Create, DoFn, MapElements}
import org.apache.beam.sdk.values.{KV, TypeDescriptors}


case class MyClass(int: Int)

class MyBinaryFn extends BinaryCombineFn[MyClass] {
  def apply(left: MyClass, right: MyClass): MyClass = {
    println(left, right)
    if (left.int > right.int) left else right
  }
}

object BinaryCombineFnMain {

  import collection.JavaConverters._

  def main(args: Array[String]): Unit = {
    val pipeline = Pipeline.create()

    pipeline
      .apply("create elems", Create.of(Seq(MyClass(1), MyClass(3), MyClass(0), MyClass(2)).asJava))
      .apply("Reduce", Combine.globally(new MyBinaryFn))
      .apply(MapElements
        .into(TypeDescriptors.strings())
        .via { (x: MyClass) =>
          println(x)
          x.toString
        }
      )

    pipeline.run()
  }

}

object BinaryCombineFnPerKeyMain {

  import collection.JavaConverters._

  def main(args: Array[String]): Unit = {
    val pipeline = Pipeline.create()

    pipeline
      .apply("create kvs", Create.of(Seq(
        KV.of(1, MyClass(1)), KV.of(1, MyClass(3)),
        KV.of(2, MyClass(2)), KV.of(2, MyClass(0))).asJava))
      .apply("ReduceByKey", Combine.perKey[Int, MyClass, MyClass](new MyBinaryFn))
      .apply(MapElements
        .into(TypeDescriptors.strings())
        .via { (x: KV[Int, MyClass]) =>
          println(x)
          x.toString
        }
      )

    pipeline.run()
  }

}

object BinaryCombineFnPerKeyWithHotKeyFanoutMain {

  import collection.JavaConverters._

  def main(args: Array[String]): Unit = {
    val pipeline = Pipeline.create()

    pipeline
      .apply("create kvs", Create.of(Seq(
        KV.of(1, MyClass(1)), KV.of(1, MyClass(3)),
        KV.of(2, MyClass(2)), KV.of(2, MyClass(0))).asJava))
      .apply("ReduceByKey", Combine
        .perKey[Int, MyClass, MyClass](new MyBinaryFn)
        .withHotKeyFanout(10))
      .apply(MapElements
        .into(TypeDescriptors.strings())
        .via { (x: KV[Int, MyClass]) =>
          println(x)
          x.toString
        }
      )

    pipeline.run()
  }

}