Hi,
what is the timescale of your data in Kafka. If you have data in there
that spans more than ~30 minutes I would expect your windows to fire
very soon after the job is started. Event time does not depend on a wall
clock but instead advances with the time in the stream. As Flink
advances through the data in Kafka so does event-time advance in step.
Does that explain your situation?
Best,
Aljoscha
On 15.06.20 16:49, Ori Popowski wrote:
I'm using Flink 1.10 on YARN, and I have a EventTimeSessionWindow with a
gap of 30 minutes.
But as soon as I start the job, events are written to the sink (I can see
them in S3) even though 30 minutes have not passed.
This is my job:
val stream = senv
.addSource(new FlinkKafkaConsumer("…", compressedEventDeserializer,
properties))
.filter(_.sessionId.nonEmpty)
.flatMap(_ match { case (_, events) => events })
.assignTimestampsAndWatermarks(new
TimestampExtractor[Event](Time.minutes(10)) {
override def extractTimestamp(element: Event): Long =
event.sequence / 1000 // microseconds
})
.keyBy(_.sessionId)
.window(EventTimeSessionWindows.withGap(Time.of(30, MINUTES)))
.process(myProcessWindowFunction)
AsyncDataStream.unorderedWait(stream, myAsyncS3Writer, 30, SECONDS, 100)
Any idea why it's happening?
