flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Compilation Error in WindowStream.fold()
Date Fri, 24 Feb 2017 17:14:20 GMT
Hi Nirmalya,
what does the compiler say if you use the variant without explicit
TypeInfo? Like this:

 .fold(
           // Seed
           Map[EWayCoordinates,Set[VehicleID]](),

           // FoldFunction
           folder,

           // WindowFunction
           windower,
  )

Best,
Aljoscha

On Thu, 23 Feb 2017 at 14:41 nsengupta <sengupta.nirmalya@gmail.com> wrote:

> For reasons I cannot grasp, I am unable to move ahead.
>
> Here's the code:
>
> ---------------------------------------------------------------------------------------------------------------------------------------------
>
>
> import org.apache.flink.api.common.functions.FoldFunction
> import org.apache.flink.api.java.typeutils.TupleTypeInfo
> import org.apache.flink.streaming.api.TimeCharacteristic
> import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction
> import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
> import org.apache.flink.streaming.api.scala._
> import
> org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
> 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
> import org.nirmalya.exercise.Elements.{EWayCoordinates, PositionReport,
> RawMITSIMTuple, VehicleID}
>
> case class EWayCoordinates(eWayID: Int, eWayDir: Int, eWayLane: Int,
> eWaySegment: Int)
>
> case class VehicleDetails(vehicleID: Int, vehicleSpeed: Int, vehiclePos:
> Int)
>
> case class PositionReport(
>                               // tupletype: Int,
>                               timeOfReport: Int,
>                               eWayCoordinates: EWayCoordinates,
>                               vehicleDetails: VehicleDetails
>                        )
>
>
> // ....
>
>
> val envDefault = StreamExecutionEnvironment.createLocalEnvironment(4)
>   envDefault
>     .setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>
>   val readings = IndexedSeq [RawMITSIMTuple] (
>     RawMITSIMTuple(0,0,107,32,0,0,0,10, 53320,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,0,109,20,0,0,0,10,100644,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,0,106,28,0,0,0,10,137745,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,1,107,32,0,0,0,67,354281,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,1,105,30,0,0,1,94,501089,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,1,110,26,0,0,1,95,495898,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,1,102,30,0,0,1,85,453562,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,2,104,32,0,0,1,80,427144,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,2,102,20,0,0,1,73,390383,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,2,111,27,1,0,1,69,369135,-1,-1,-1,-1,-1,-1),
>     RawMITSIMTuple(0,2,112,28,1,0,0, 1,  5757,-1,-1,-1,-1,-1,-1)
>   )
>
> val folder = new FoldFunction[PositionReport,
> Map[EWayCoordinates,Set[Int]]]
> {
>       override
>         def fold(
>               t: Map[EWayCoordinates, Set[VehicleID]], o: PositionReport
>         ): Map[EWayCoordinates, Set[VehicleID]] = {
>         t + (o.eWayCoordinates -> (t.getOrElse(o.eWayCoordinates,Set.empty)
> + (o.vehicleDetails.vehicleID)))
>       }
>     }
>
>     val windower = new AllWindowFunction[Map[EWayCoordinates,
> Set[VehicleID]],(EWayCoordinates,Int),Window] {
>       override
>       def apply(
>            w: Window,
>            bunch: Iterable[Map[EWayCoordinates, Set[VehicleID]]],
>            collector: Collector[(EWayCoordinates, VehicleID)]): Unit = {
>
>         val allVehiclesInLast30Mins = bunch.iterator().next.mapValues(e =>
> e.size)
>
>         allVehiclesInLast30Mins.foreach(e => println(e))
>
>         collector.collect((EWayCoordinates(-1,-1,-1,-1),0))
>
>       }
>     }
>
>     val uniqueVehicles = envDefault
>       .fromCollection(readings)
>       .map(e => MITSIMUtils.preparePositionReport(e))
>       .assignAscendingTimestamps(e => e.timeOfReport)
>       .keyBy(e => (
>         e.eWayCoordinates.eWayID,
>         e.eWayCoordinates.eWayDir,
>         e.eWayCoordinates.eWaySegment,
>         e.vehicleDetails.vehicleID))
>       .windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
>       .fold(
>            // Seed
>            Map[EWayCoordinates,Set[VehicleID]](),
>
>            // FoldFunction
>            folder,
>
>            // WindowFunction
>            windower,
>
>            // Satisfying the compiler
>            new TupleTypeInfo[Map[EWayCoordinates,Set[VehicleID]]](),
>            new TupleTypeInfo[(EWayCoordinates,Int)]
>       )
>
>
> -----------------------------------------------------------------------------------------
>
> The compiler is unhappy:
>
> [ERROR]
>
> /home/nirmalya/Workspace-Flink/LinearRoad/src/test/scala/org/nirmalya/exercise/StreamStateTest.scala:136:
> error: missing argument list for method fold in class AllWindowedStream
> [ERROR] Unapplied methods are only converted to functions when a function
> type is expected.
> [ERROR] You can make this conversion explicit by writing `fold _` or
> `fold(_)(_)(_)` instead of `fold`.
> [ERROR]       .fold(
> [ERROR]            ^
> [ERROR] one error found
>
>
> ----------------------------------------------------------------------------------------
>
> I understand why is the compiler unhappy, but I am unsure if I have to go
> through all the *devilry*. In no Flink example, I see some such thing being
> prescribed. But, then, perhaps I am missing an important point.
>
> I have been through this  comment
> <
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Incremental-aggregations-Example-not-working-td10581.html#a10585
> >
> by *Yassine Marzougui*, before I added those type hints. But, I am using
> *Flink 1.2.0*.
>
> I know this sounds silly, but I am simply failing to get out of this.
>
> All help appreciated.
>
> -- Nirmalya
>
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Compilation-Error-in-WindowStream-fold-tp11830.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Mime
View raw message