Apache FlinkのFlatMapFunctionで型を忠実に派生させるにはどうすればいいですか？

Question

ScalaでFlatMapFunctionクラスを使用しているときに、型を暗黙的に指定するにはどうすればよいですか？

val test = env.fromElements((0,1111),(1,2222)) 

test.flatMap(new FlatMapFunction[A,B]() { 
     override def flatMap(implicit x:A, out:Collector[B]):Unit = { 
     x => x match { 
      case (k,v) => 
      if (moveToP(v)) 
       out.write(v) 
     } 
     } 
    }) 

エラー：trait FlatMapFunctionは型パラメータをとります。

Answer 1

あなたは何を求めているのですが、とにかく答えるつもりです。

ジェネリック型でオブジェクトをインスタンス化することはできません。ストリームのタイプがDataStream[(Int, Int)]の場合、new FlatMapFunction[(Int, Int), Int]を適用する必要があります。 Scalaで

あなたはflatMapのためのラムダ式を使用することができます

test.flatMap((x, out: Collector[Int]) => { 
    x match { 
    case (k,v) => 
     if (moveToP(v)) 
     out.collect(v) 
    } 
}).print() 

やコレクターなしを：

test.flatMap(x => { 
    if (moveToP(x._2)) List(x._2) else List.empty 
})