Функции API Flink Scala по общим параметрам

Question

Это вопрос следующего содержания: Flink Scala API "not enough arguments".

Хотелось бы пройти мимо Flink's DataSet и сделать что-то с ним, но параметры набора данных являются общими.

Вот проблема у меня сейчас:

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) 
} 

У меня есть эта ошибка для ds.map(r => 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] 

Конечно, id функция здесь только один пример, и я хотел бы быть в состоянии сделать с ней что-то более сложное.

Как это можно решить?

Answer 1

вы также должны иметь TypeInformation в качестве контекста, связанного с параметром К, так:

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

Причина заключается в том, что Flink анализирует типы, которые вы используете в своей программе, и создает экземпляр TypeInformation для каждого тип, который вы используете. Если вы хотите создать общие операции, вам нужно убедиться, что TypeInformation этого типа доступен, добавив привязку к контексту. Таким образом, компилятор Scala убедится, что экземпляр доступен на сайте вызова общей функции.