flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Flink 1.1 event-time windowing changes from 1.0.3
Date Mon, 08 Aug 2016 12:01:42 GMT
Hi Adam,
sorry for the inconvenience. This is caused by a new file read operator,
specifically how it treats watermarks/timestamps. I opened an issue here
that describes the situation:
https://issues.apache.org/jira/browse/FLINK-4329.

I think this should be fixed for an upcoming 1.1.1 bug fixing release.

Cheers,
Aljoscha

On Sat, 6 Aug 2016 at 12:33 Adam Warski <adam@warski.org> wrote:

> Hello,
>
> I have a very simple stream where I window data using event-time.
> As a data source I’m using a CSV file, sorted by increasing timestamps.
>
> Here’s the source:
>
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>
> val lines = env.readTextFile(csvFileName)
>
> lines
>   .flatMap { l => parseLine(l) }
>   .assignAscendingTimestamps(t => t.timestampSeconds * 1000L)
>   .keyBy(t => t.key)
>   .timeWindow(Time.minutes(30), Time.minutes(5))
>   .fold(0)((c, _) => c+1)
>   .addSink { c =>
>     println(c)
>   }
>
> env.execute()
>
> This used to work fine in 1.0.3, that is the aggregate counts are printed
> to stdout.
>
> However after updating to 1.1, nothing happens - I can see the stages
> being initialized (switching state from SCHEDULED to DEPLOYING to RUNNING),
> but then immediately going to FINISHED, without printing anything out.
>
> If I add a .map {x => println(x); x} after .assignAscendingTimestamps I
> can see the data flowing - so data *is* being read, just somehow the
> windowing causes it to be lost?
>
> Any ideas on where to look for possible causes?
>
> Thanks!
>
> --
> Adam Warski
>
> http://twitter.com/#!/adamwarski
> http://www.softwaremill.com
> http://www.warski.org
>
>

Mime
View raw message