Apache Beam / Scala 组合 perkey

Question

我正在找人提供一个使用 Apache Beam Combine.perkey 转换在 Scala 中使用 beam Java API 的工作示例。

我遇到了 scala/java 类型互操作性的问题。

我无法让每个键组合起来工作。我永远无法让它在语法上是正确的。

例子：

val sales: PCollection KV[(Int,Int), Long]

sales.apply(Combine.perKey[(Int,Int),Long,Long](new SumLongs()))

import org.apache.beam.sdk.transforms.SerializableFunction

class SumLongs extends SerializableFunction[Iterable[Long], Long] {
  override def apply(input: Iterable[Long]): Long = {
  var sum = 0L
  for (item <- input) {
    sum += item
  }
  sum
  }
}

它给出错误，“perkey 的类型参数太多”。当我把它拿出来时，它会显示“未指定的类型参数：OutputT”。

Answer 1

我只需要用 java.lang.Iterable 更改 scala Iterable