FlinkScalaAPI在泛型参数上执行函数-FlinkScalaAPIfunctionsongenericparameters

It's a follow up question on Flink Scala API "not enough arguments".

这是关于Flink Scala API“参数不足”的后续问题。

I'd like to be able to pass Flink's DataSets around and do something with it, but the parameters to the dataset are generic.

我希望能够传递Flink的数据集并对其进行处理，但是数据集的参数是通用的。

Here's the problem I have now:

我现在遇到的问题是:

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
import scala.reflect.ClassTag

object TestFlink {

  def main(args: Array[String]) {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val text = env.fromElements(
      "Who's there?",
      "I think I hear them. Stand, ho! Who's there?")

    val split = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
    id(split).print()

    env.execute()
  }

  def id[K: ClassTag](ds: DataSet[K]): DataSet[K] = ds.map(r => r)
}

I have this error for ds.map(r => r):

这是ds的误差。地图(r = >):

Multiple markers at this line
    - not enough arguments for method map: (implicit evidence$256: org.apache.flink.api.common.typeinfo.TypeInformation[K], implicit 
     evidence$257: scala.reflect.ClassTag[K])org.apache.flink.api.scala.DataSet[K]. Unspecified value parameters evidence$256, evidence$257.
    - not enough arguments for method map: (implicit evidence$4: org.apache.flink.api.common.typeinfo.TypeInformation[K], implicit evidence
     $5: scala.reflect.ClassTag[K])org.apache.flink.api.scala.DataSet[K]. Unspecified value parameters evidence$4, evidence$5.
    - could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[K]

Of course, the id function here is just an example, and I'd like to be able to do something more complex with it.

当然，这里的id函数只是一个例子，我想用它做一些更复杂的事情。

How it can be solved?

如何解决这个问题?

1 个解决方案

def id[K: ClassTag: TypeInformation](ds: DataSet[K]): DataSet[K] = ds.map(r => r)