Teodor,
This is happening because of the way that readTextFile works when it is
executing in parallel, which is to divide the input file into a bunch of
splits, which are consumed in parallel. This is making it so that the
watermark isn't able to move forward until much or perhaps all of the file
has been read. If you change the parallelism of the source to 1, like this
final DataStream<String> linesIn =
env.readTextFile(fileNameInput).setParallelism(1);
then you should see the job make steady forward progress with windows
closing on a regular basis.
Regards,
David
On Sat, Aug 29, 2020 at 4:59 PM Teodor Spæren <teodor_spae...@riseup.net>
wrote:
> Hey!
>
> Second time posting to a mailing lists, lets hope I'm doing this
> correctly :)
>
> My usecase is to take data from the mediawiki dumps and stream it into
> Flink via the `readTextFile` method. The dumps are TSV files with an
> event per line, each event have a timestamp and a type. I want to use
> event time processing and simply print out how many of each event type
> there is per hour. The data can be out of order, so I have 1 hour
> tolerance.
>
> What I expect to happen here is that as it goes through a month of data,
> it will print out the hours as the watermark passes 1 hour. So I'll get
> output continuously until the end.
>
> What really happens is that the program outputs nothing until it is done
> and then it outputs everything. The timestamp is also stuck at
> 9223372036854776000 in the web management. If I switch to using
> CountWindows instead of timewindows, it outputs continuously like I
> would expect it too, so it seems to be watermark related.
>
> I'm running Flink version 1.11.1 on JVM version:
>
> OpenJDK 64-Bit Server VM - GraalVM Community - 11/11.0.7+10-jvmci-20.1-b02
>
> The parallel setting is 1 and it's running on my laptop.
>
>
> I don't know how much code I'm allowed to attach here, so I've created a
> github repo with the complete self standing example [1]. To get the data
> used, run the following commands:
>
> $ wget
> https://dumps.wikimedia.org/other/mediawiki_history/2020-07/enwiki/2020-07.enwiki.2016-04.tsv.bz2
> $ pv -cN source < 2020-07.enwiki.2016-04.tsv.bz2 | bzcat | pv -cN bzcat
> | sort -k4 > 2020-07.enwiki.2016-04.sorted.tsv
>
> If you don't have pv installed, just remove that part, I just like to
> have an overview.
>
>
> The main code part is this:
>
> package org.example.prow;
>
> import org.apache.flink.api.common.eventtime.WatermarkStrategy;
> import org.apache.flink.streaming.api.TimeCharacteristic;
> import org.apache.flink.streaming.api.datastream.DataStream;
> import org.apache.flink.streaming.api.datastream.KeyedStream;
> import
> org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
> import
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import
> org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
> import org.apache.flink.streaming.api.windowing.time.Time;
> import org.example.prow.wikimedia.Event;
>
> import java.time.Duration;
>
> public class App {
> public static void main(String[] args) throws Exception {
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>
> final String fileNameInput =
> "file:///home/rhermes/madsci/thesis/data/mediawiki_history/2020-07.enwiki.2016-04.sorted.tsv";
> final DataStream<String> linesIn =
> env.readTextFile(fileNameInput);
>
>
> final SingleOutputStreamOperator<Event> jj = linesIn.map(value ->
> new Event(value));
>
> final WatermarkStrategy<Event> mew =
> WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofHours(1)).withTimestampAssigner((element,
> recordTimestamp) -> element.eventTimestamp.toEpochSecond() * 1000);
>
> final DataStream<Event> props =
> jj.assignTimestampsAndWatermarks(mew);
>
> final KeyedStream<Event, String> praps = props.keyBy(e ->
> e.eventEntity.toString());
>
>
>
> praps.window(TumblingEventTimeWindows.of(Time.hours(1))).sum("something").print("JAJ!");
>
> env.execute("FlinkWikipediaHistoryTopEditors");
> }
> }
>
> If you see any erors here, please tell me, this is sort of driving me
> mad >_<.
>
> Best regards,
> Teodor Spæren
>
> [1] https://github.com/rHermes/flink-question-001
>
