flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Scala case classes with a generic parameter
Date Thu, 09 Jun 2016 08:50:17 GMT
Hi James,
the TypeInformation must be available at the call site, not in the case
class definition. In your WindowFunction you are using a TestGen[String] so
it should suffice to add this line at some point before the call to apply():

implicit val testGenType = createTypeInformation[TestGen[String]]

Hope that helps.

Best,
Aljoscha

On Wed, 1 Jun 2016 at 20:11 James Bucher <jbucher@expedia.com> wrote:

> Hi,
>
> I have been trying to get a case class with a generic parameter working
> with Filnk 1.0.3 and have been having some trouble. However when I compile
> I get the following error:
> debug-type-bug/src/main/scala/com/example/flink/jobs/CaseClassWithGeneric.scala:40:
> error: could not find implicit value for evidence parameter of type
> org.apache.flink.api.common.typeinfo.TypeInformation[com.example.flink.jobs.CaseClassWithGeneric.TestGen[String]]
> [ERROR]           .apply(new AggregateOrigins)
>
> I am importing org.apache.flink.api.scala._ and the generic type is
> defined as [T: TypeInformation] as suggested here:
> https://ci.apache.org/projects/flink/flink-docs-master/internals/types_serialization.html
>
>
> The full code for the program is as follows:
>
> package com.example.flink.jobs
>
> import java.util.{Properties}
> import org.apache.flink.api.common.typeinfo.TypeInformation
> import org.apache.flink.streaming.util.serialization.SimpleStringSchema
> import org.apache.flink.util.Collector
> import org.apache.flink.streaming.api.scala.function.WindowFunction
> import org.apache.flink.streaming.api.windowing.time.Time
> import org.apache.flink.streaming.api.windowing.windows.TimeWindow
> import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
> import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer08}
> import org.apache.flink.api.scala._
>
> object CaseClassWithGeneric {
>   case class TestGen[T: TypeInformation](item: T) {}
>
>   class AggregateOrigins extends WindowFunction[String, TestGen[String], String, TimeWindow]
{
>     def apply(key: String, win: TimeWindow, values: Iterable[String], col: Collector[TestGen[String]]):
Unit = {
>       values.foreach(x => { })
>       col.collect(new TestGen[String]("Foo"))
>     }
>   }
>
>   def main(args: Array[String]): Unit = {
>     val env = StreamExecutionEnvironment.getExecutionEnvironment
>     val properties = new Properties();
>     val messageStream = env.addSource(
>       new FlinkKafkaConsumer08("topic", new SimpleStringSchema, properties))
>           .keyBy(s => s)
>           .timeWindow(Time.days(1))
>           .apply(new AggregateOrigins)
>     messageStream.print()
>     env.execute("Simple Job")
>   }
> }
>
> When I dug into the apply() function definition I found the following:
>
> def apply[R: TypeInformation](
>     function: WindowFunction[T, R, K, W]): DataStream[R] = {
>
>   val cleanFunction = clean(function)
>   val applyFunction = new ScalaWindowFunctionWrapper[T, R, K, W](cleanFunction)
>   asScalaStream(javaStream.apply(applyFunction, implicitly[TypeInformation[R]]))
> }
>
> As Far as I can tell TestGen[String] should correspond to [R: TypeInformation] in apply.
Am I missing something or is it not possible to define case class with a generic parameter?
>
> Thanks,
>
> James Bucher
>
>

Mime
View raw message