could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[...]

Question

I am trying to write some use cases for Apache Flink. One error I run into pretty often is

could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[SomeType]

My problem is that I cant really nail down when they happen and when they dont.

The most recent example of this would be the following

...
val largeJoinDataGen = new LargeJoinDataGen(dataSetSize, dataGen, hitRatio)
val see = StreamExecutionEnvironment.getExecutionEnvironment
val newStreamInput = see.addSource(largeJoinDataGen)
...

where LargeJoinDataGen extends GeneratorSource[(Int, String)] and GeneratorSource[T] extends SourceFunction[T], both defined in separate files.

When trying to build this I get

Error:(22, 39) could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[(Int, String)]
val newStreamInput = see.addSource(largeJoinDataGen)

1. Why is there an error in the given example?

2. What would be a general guideline when these errors happen and how to avoid them in the future?

P.S.: first scala project and first flink project so please be patient

Answer 1

This mostly happens when you have user code, i.e. a source or a map function or something of that nature that has a generic parameter. In most cases you can fix that by adding something like

implicit val typeInfo = TypeInformation.of(classOf[(Int, String)])

If your code is inside another method that has a generic parameter you can also try adding a context bound to the generic parameter of the method, as in

def myMethod[T: TypeInformation](input: DataStream[Int]): DataStream[T] = ...

Answer 2

You may make an import instead of implicits

import org.apache.flink.streaming.api.scala._

It will also help.

Answer 3

My problem is that I cant really nail down when they happen and when they dont.

They happen when an implicit parameter is required. If we look at the method definition we see:

def addSource[T: TypeInformation](function: SourceFunction[T]): DataStream[T]

But we don't see any implicit parameter defined, where is it?

When you see a polymorphic method where the type parameter is of the form

def foo[T : M](param: T)

Where T is the type parameter and M is a context bound. It means that the creator of the method is requesting an implicit parameter of type M[T]. It is equivalent to:

def foo[T](param: T)(implicit ev: M[T])

In the case of your method, it is actually expanded to:

def addSource[T](function: SourceFunction[T])(implicit evidence: TypeInformation[T]): DataStream[T]

This is why you see the compiler complaining, as it can't find the implicit parameter the method is requiring.

If we go to the Apache Flink Wiki, under Type Information we can see why this happens :

No Implicit Value for Evidence Parameter Error

In the case where TypeInformation could not be created, programs fail to compile with an error stating “could not find implicit value for evidence parameter of type TypeInformation”. A frequent reason if that the code that generates the TypeInformation has not been imported. Make sure to import the entire flink.api.scala package. import org.apache.flink.api.scala._

For generic methods, you'll need to require them to generate a TypeInformation at the call-site as well:

For generic methods, the data types of the function parameters and return type may not be the same for every call and are not known at the site where the method is defined. The code above will result in an error that not enough implicit evidence is available. In such cases, the type information has to be generated at the invocation site and passed to the method. Scala offers implicit parameters for that.

Note that import org.apache.flink.streaming.api.scala._ may also be necessary.

For your types this means that if the invoking method is generic, it also needs to request the context bound for it's type parameter.