Flink Scala API functions on generic parameters

Question

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

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.

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

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.

How it can be solved?

Joey · Accepted Answer · 2015-04-22 14:51:23Z

7

you also need to have TypeInformation as a context bound on the K parameter, so:

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

The reason is, that Flink analyses the types that you use in your program and creates a TypeInformation instance for each type you use. If you want to create generic operations then you need to make sure a TypeInformation of that type is available by adding a context bound. This way, the Scala compiler will make sure an instance is available at the call site of the generic function.

edited Apr 22, 2015 at 14:51

Joey

1,34914 silver badges26 bronze badges

answered Apr 9, 2015 at 15:54

aljoscha

9865 silver badges7 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Flink Scala API functions on generic parameters

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related