Hi Sendoh,
from your description it's really hard to figure out what the problem could
be. The first thing to do would be check how many records you actually
consume from Kafka and how many items are outputted. Next I would take a
look at the timestamp extractor. Can it be that records are discarded
because they have a wrong timestamp? It could be that the elements arrive
out of order.
Cheers,
Till
On Mon, Nov 7, 2016 at 4:15 PM, Sendoh <unicorn.bana...@gmail.com> wrote:
> Hi Flink users,
>
> we have an issue to see all events in the window apply() function, while we
> see them before the window operation.
>
> The input is from Kafka and contains at least 6 topics which is at least 30
> GB in total, and we have tested locally in IDE and cluster using 1.1.3 and
> 1.0.3.
>
> It works when:
> - using version 1.0.3 we can see all events but we lost around 1/3 of them
> - using processing time
> - setting the Kafka offset as latest
> - using less topics (3 topics)
>
> But we want to use offset as earliest in event time, and read 6 topics or
> more.
>
> Our test code:
>
> List<String> topicList = getAllEventTopic(eventsConf);
>
> // Start Flink job
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>
> Properties properties = new Properties();
> properties.setProperty("bootstrap.servers",
> Config.bootstrapServers);
> properties.setProperty("group.id",
> parameter.getRequired("groupId"));
> // It works
> properties.setProperty("auto.offset.reset", "earliest");
>
> DataStream<String> streams = env.addSource(
> new FlinkKafkaConsumer09<>(topicList, new
> SimpleStringSchema(), properties));
>
> DataStream<JSONObject> jsonStreams = streams.flatMap(new
> JSONMap());
>
> // we actually use keyBy(). This is for testing purpose.
> jsonStreams.assignTimestampsAndWatermarks(new
> TestWatermark()).rebalance().windowAll(TumblingEventTimeWindows.of(
> Time.minutes(1)))
> .allowedLateness(Time.seconds(10))
> .apply(new AllWindowFunction<JSONObject, Object, TimeWindow>() {
> @Override
> public void apply(TimeWindow timeWindow, Iterable<JSONObject>
> iterable, Collector collector) throws Exception {
> Iterator<JSONObject> it = iterable.iterator();
> while(it.hasNext()){
> collector.collect(it.next());
> }
> }
> }).writeAsText("test",
> FileSystem.WriteMode.OVERWRITE).setParallelism(1);
>
> Is there any suggestion that we could try to fix the issue?
>
> Best,
>
> Sendoh
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/Cannot-see-all-
> events-in-window-apply-for-big-input-tp9945.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>
