flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "David B. Ciar" <dcia...@ceh.ac.uk>
Subject Re: Window Functions with Incremental Aggregation
Date Tue, 02 Aug 2016 12:09:36 GMT
Hello again,

Having had another go at this today, I clearly see that I cannot pass a
certain type into the fold/window function and expect to be able to return a
datastream of another type from the window function.  I have tried a
different approach and am now receiving a run-time exception, caused by
trying to use a composite case class as the fold accumulator value.  My
query now is whether this is possible, and if it is possible, how to fix the
run-time exception.  Again any help is appreciated.

The exception:

Exception in thread "main" java.lang.ClassCastException: [Ljava.lang.Object;
cannot be cast to [Lorg.apache.flink.api.common.typeinfo.TypeInformation;
	at
org.management.observations.processing.jobs.QCBlockNull$$anon$6.<init>(QCBlockNull.scala:104)
	at
org.management.observations.processing.jobs.QCBlockNull$.main(QCBlockNull.scala:104)
	at
org.management.observations.processing.jobs.QCBlockNull.main(QCBlockNull.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)

The code snippet is pasted below, but also neater formatted Gist link:

// The cause of the exception is the .apply(...) below and the use of
IncrementalPlaceHolder.  The fold and window classes return type
IncrementalWindowPlaceholder

    val nullQCEvents1h = nullStream
      .keyBy("feature","procedure")
      .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(30)))
      .apply(new IncrementalWindowPlaceholder(0,None,None,None),
        new QCFoldCounter(),
        new QCCheckNullAggregate())

// The aggregate class I want to use with the fold/window function and emit
as the DataStream type:

case class IncrementalWindowPlaceholder (foldedValue: Double,
                                         keys: Option[Tuple],
                                         startTime: Option[Long],
                                         endTime: Option[Long]){

  override def toString: String =
   
foldedValue.toString+','+keys.getOrElse('-')+','+startTime.getOrElse('-')+','+endTime.getOrElse('-')
}

Also here:
https://gist.github.com/dbciar/904e2d35d6aae30214666de1176f1d7c





--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Window-Functions-with-Incremental-Aggregation-tp8246p8259.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.

Mime
View raw message