[jira] [Commented] (FLINK-10167) SessionWindows not compatible with typed DataStreams in scala

["Aljoscha Krettek (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-10167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16587179#comment-16587179 ] 

Aljoscha Krettek commented on FLINK-10167:
------------------------------------------

Yes, your right that this is a problem but only when working with "values". If you change it to this it will work:
{code}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.{TimeWindow, Window}
import org.apache.flink.util.Collector

object TestJob {
  val jobName = "TestJob"

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.fromCollection(Range(0, 100).toList)
      .map( in => (in / 10, in) )
      .keyBy(_._1)
      .window(ProcessingTimeSessionWindows.withGap(Time.minutes(1)))
      .reduce(
        (a: (Int, Int), b: (Int, Int)) => (a._1, a._2 + a._2) ,
        (key: Int, window: TimeWindow, items: Iterable[(Int, Int)], out: Collector[String]) => s"${key}: ${items}"
      )
      .map(println(_))

    env.execute(jobName)
  }
}
{code}

> SessionWindows not compatible with typed DataStreams in scala
> -------------------------------------------------------------
>
>                 Key: FLINK-10167
>                 URL: https://issues.apache.org/jira/browse/FLINK-10167
>             Project: Flink
>          Issue Type: Bug
>            Reporter: Andrew Roberts
>            Priority: Major
>
> I'm trying to construct a trivial job that uses session windows, and it looks like the data type parameter is hardcoded to `Object`/`AnyRef`. Due to the invariance of java classes in scala, this means that we can't use the provided SessionWindow helper classes in scala on typed streams.
>  
> Example job:
> {code:java}
> import org.apache.flink.api.scala._
> import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
> import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
> import org.apache.flink.streaming.api.windowing.time.Time
> import org.apache.flink.streaming.api.windowing.windows.{TimeWindow, Window}
> import org.apache.flink.util.Collector
> object TestJob {
>   val jobName = "TestJob"
>   def main(args: Array[String]): Unit = {
>     val env = StreamExecutionEnvironment.getExecutionEnvironment
>     env.fromCollection(Range(0, 100).toList)
>       .keyBy(_ / 10)
>       .window(ProcessingTimeSessionWindows.withGap(Time.minutes(1)))
>       .reduce(
>         (a: Int, b: Int) => a + b,
>         (key: Int, window: Window, items: Iterable[Int], out: Collector[String]) => s"${key}: ${items}"
>       )
>       .map(println(_))
>     env.execute(jobName)
>   }
> }{code}
>  
> Compile error:
> {code:java}
> [error]  found   : org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
> [error]  required: org.apache.flink.streaming.api.windowing.assigners.WindowAssigner[_ >: Int, ?]
> [error] Note: Object <: Any (and org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows <: org.apache.flink.streaming.api.windowing.assigners.MergingWindowAssigner[Object,org.apache.flink.streaming.api.windowing.windows.TimeWindow]), but Java-defined class WindowAssigner is invariant in type T.
> [error] You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
> [error]       .window(ProcessingTimeSessionWindows.withGap(Time.minutes(1))){code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)