You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Fabian Hueske <fh...@gmail.com> on 2019/09/02 13:52:21 UTC
Re: End of Window Marker
Hi Padarn,
Regarding your throughput concerns: A sink task may write to multiple
partitions, but each partition may only be written by a single task.
@Eduardo: Thanks for sharing your approach! Not sure if I understood it
correctly, but I think that the approach does not guarantee that all
results of a window are emitted before the end-of-window marker is written.
Since the sink operator and the single-task-operator are separate
operators, the output records might get stuck (or be bufffered) in one of
the sink tasks and the single-task would still emit an end-of-window marker
record because it doesn't know about the sink task.
Best,
Fabian
Am Do., 29. Aug. 2019 um 18:42 Uhr schrieb Eduardo Winpenny Tejedor <
eduardo.winpenny@gmail.com>:
> Hi,
>
> I'll chip in with an approach I'm trying at the moment that seems to work,
> and I say seems because I'm only running this on a personal project.
>
> Personally, I don't have anything against end-of-message markers per
> partition, Padarn you seem to not prefer this option as it overloads the
> meaning of the output payload. My approach is equally valid when producing
> watermarks/end-of-message markers on a side output though.
>
> The main problem of both approaches is knowing when the window has
> finished across all partitions without having to wait for the start of the
> next window.
>
> I've taken the approach of sending all output messages of the window to 1.
> the sink but also 2. a single task operator. The single task operator
> registers an event time based timer at the time of the end of the window.
> You have the confidence of the task's timer triggering only once at the
> right time because all the post-window watermarks go through to the same
> task. At that point I make the task send an end-of-message marker to every
> partition. I don't need to send the count because Kafka messages are
> ordered. AND IF you prefer to not overload the semantic of your original
> Kafka topic you can post the message to a separate location of your choice.
>
> While this does mean that the end of marker message only gets sent through
> once the window has finished across all substreams (as opposed to per
> stream), it does mean you don't need to wait for the next window to start
> AND the watermark gap between substreams should never grow that much anyway.
>
> This approach should be particularly useful when the number of partitions
> or keying mechanism is different between the input and output topics.
>
> Hopefully that doesn't sound like a terrible idea.
>
> eduardo
>
>
>
>
> On Wed, 28 Aug 2019, 02:54 Padarn Wilson, <pa...@gmail.com> wrote:
>
>> Hi again Fabian,
>>
>> Thanks for pointing this out to me. In my case there is no need for keyed
>> writing - but I do wonder if having each kafka task write only to a single
>> partition would significantly affect performance.
>>
>> Actually now that I think about it, the approach to just wait for the
>> first records of the next window is also subject to the problem you mention
>> above: a producer lagging behind the rest could end up with a partition
>> containing element out of ‘window order’.
>>
>> I was also thinking this problem is very similar to that of checkpoint
>> barriers. I intended to dig into the details of the exactly once Kafka sink
>> for some inspiration.
>>
>> Padarn
>>
>> On Tue, 27 Aug 2019 at 11:01 PM, Fabian Hueske <fh...@gmail.com> wrote:
>>
>>> Hi Padarn,
>>>
>>> Yes, this is quite tricky.
>>> The "problem" with watermarks is that you need to consider how you write
>>> to Kafka.
>>> If your Kafka sink writes to keyed Kafka stream (each Kafka partition is
>>> written by multiple producers), you need to broadcast the watermarks to
>>> each partition, i.e., each partition would receive watermarks from each
>>> parallel sink task. So in order to reason about the current watermark of a
>>> partition, you need to observe them and take the minimum WM across all
>>> current sink task WMs.
>>> Things become much easier, if each partition is only written by a single
>>> task but this also means that data is not key-partitioned in Kafka.
>>> In that case, the sink task only needs to write a WM message to each of
>>> its assigned partitions.
>>>
>>> Hope this helps,
>>> Fabian
>>>
>>>
>>> Am Sa., 17. Aug. 2019 um 05:48 Uhr schrieb Padarn Wilson <
>>> padarn@gmail.com>:
>>>
>>>> Hi Fabian, thanks for your input
>>>>
>>>> Exactly. Actually my first instinct was to see if it was possible to
>>>> publish the watermarks somehow - my initial idea was to insert regular
>>>> watermark messages into each partition of the stream, but exposing this
>>>> seemed quite troublesome.
>>>>
>>>> > In that case, you could have a ProcessFunction that is chained before
>>>> the sink and which counts the window results per time slice and emits the
>>>> result when the watermark passes to a side output.
>>>> All side output messages are collected by a single task and can be
>>>> published to a Kafka topic or even be made available via Queryable State.
>>>>
>>>> I understand the idea here (and exactly once semantics are probably
>>>> fine for my use case), but counting events seems a bit fragile. I'm not
>>>> totally confident the consumer can guarantee it won't read duplicates (its
>>>> a golang kafka library that seems to have some quirks).
>>>>
>>>> I think ideally each partition of the kafka topic would have some
>>>> regular information about watermarks. Perhaps the kafka producer can be
>>>> modified to support this.
>>>>
>>>> Padarn
>>>>
>>>> On Fri, Aug 16, 2019 at 3:50 PM Fabian Hueske <fh...@gmail.com>
>>>> wrote:
>>>>
>>>>> Hi Padarn,
>>>>>
>>>>> What you describe is essentially publishing Flink's watermarks to an
>>>>> outside system.
>>>>> Flink processes time windows, by waiting for a watermark that's past
>>>>> the window end time. When it receives such a WM it processes and emits all
>>>>> ended windows and forwards the watermark.
>>>>> When a sink received a WM for say 12:45:15, you know that all window
>>>>> results with until 12:45:00 have been emitted.
>>>>> Hence, the watermark tells you about the completeness of data.
>>>>>
>>>>> However, using this information is not so easy, mostly because of the
>>>>> failure semantics.
>>>>> Things become much easier if you produce to Kafka with exactly-once
>>>>> semantics.
>>>>>
>>>>> In that case, you could have a ProcessFunction that is chained before
>>>>> the sink and which counts the window results per time slice and emits the
>>>>> result when the watermark passes to a side output.
>>>>> All side output messages are collected by a single task and can be
>>>>> published to a Kafka topic or even be made available via Queryable State.
>>>>>
>>>>> For at-least once output, it's much harder because you'll have
>>>>> duplicates in the output after a job recovered.
>>>>>
>>>>> Best, Fabian
>>>>>
>>>>> I think you have two options to let the consuming app know about the
>>>>> progress.
>>>>> You can either
>>>>>
>>>>> The ProcessFunction could count per window end timestamp how many
>>>>> records passed and forward that information via a side output.
>>>>> You could then
>>>>>
>>>>>
>>>>> Essentially, you'd like to publish Flink's watermarks to an outside
>>>>> system (possibly via Kafka).
>>>>>
>>>>>
>>>>> Am Mo., 12. Aug. 2019 um 14:33 Uhr schrieb Padarn Wilson <
>>>>> padarn@gmail.com>:
>>>>>
>>>>>> Hello Users,
>>>>>>
>>>>>> I have a question that is perhaps not best solved within Flink: It
>>>>>> has to do with notifying a downstream application that a Flink window has
>>>>>> completed.
>>>>>>
>>>>>> The (simplified) scenario is this:
>>>>>> - We have a Flink job that consumes from Kafka, does some
>>>>>> preprocessing, and then has a sliding window of 10 minutes and slide time
>>>>>> of 1 minute.
>>>>>> - The number of keys in each slide is not fixed
>>>>>> - The output of the window is then output to Kafka, which is read by
>>>>>> a downstream application.
>>>>>>
>>>>>> What I want to achieve is that the downstream application can someone
>>>>>> know when it has read all of the data for a single window, without waiting
>>>>>> for the next window to arrive.
>>>>>>
>>>>>> Some options I've considered:
>>>>>> - Producing a second window over the window results that counts the
>>>>>> output size, which can then be used by the downstream application to see
>>>>>> when it has received the same number: This seems fragile, as there it
>>>>>> relies on there being no loss or duplication of data. Its also an extra
>>>>>> window and Kafka stream which is a tad messy.
>>>>>> - Somehow adding an 'end of window' element to each partitions of the
>>>>>> Kafka topic which can be read by the consumer: This seems a bit messy
>>>>>> because it mixes different types of events into the same Kafka stream, and
>>>>>> there is no really simple way to do this in Flink
>>>>>> - Package the whole window output into a single message and make this
>>>>>> the unit of transaction: This is possible, but the message would be quite
>>>>>> large then (at least 10s of mb), as the volume of this stream is quite
>>>>>> large.
>>>>>> - Assume that if the consumer has no elements to read, or if the next
>>>>>> window has started to be read, then it has read the whole window: This
>>>>>> seems reasonable, and if it wasn't for the fact that my consumer on the
>>>>>> application end was a bit inflexible right now, it is probably the solution
>>>>>> I would use.
>>>>>>
>>>>>> Any further/better ideas?
>>>>>>
>>>>>> Thanks
>>>>>> Padarn
>>>>>>
>>>>>
Re: End of Window Marker
Posted by Eduardo Winpenny Tejedor <ed...@gmail.com>.
Hi all,
I'll illustrate my approach with an example as it is definitely
unorthodox. Here's some sample code. It works for me...I hope there
are no (obvious) flaws!
//myStream should be a stream of objects associated to a timestamp.
the idea is to create a Flink app that
//sends each object to kafka with the ability of also sending an extra
end-of-stream message after all events
//for the same associated timestamp have been sent
final SingleOutputStreamOperator<Tuple2<Long, Object>> myStream =
null; //replace with actual stream
final String KAFKA_SINK_TOPIC = "OUTPUT_TOPIC";
final SinkFunction<Tuple2<Long, Object>> singleObjectKafkaSink = null;
//replace with...
//needs to take care of mapping a timestamp into an end-of-stream marker record
final KeyedSerializationSchema serializationSchema = null; //replace with...
final Properties producerProperties = null; //replace with...
final int kafkaProducerPoolSize = 6; //for example...
//sinks every event of a window
myStream.addSink(singleObjectKafkaSink);
//sends one end-of-stream message per kafka partition
myStream
.map(item -> item._1()) //keep only the time
.keyBy(item -> 1) //forces every event to go to the same task
.process(new KeyedProcessFunction<Integer, Long, Tuple2<Long,
Integer>>() {
private KafkaProducer<String, Object> kafkaProducer;
@Override
public void open(Configuration parameters) throws Exception {
kafkaProducer = new KafkaProducer<>(producerProperties);
}
@Override
public void close() throws Exception {
if (kafkaProducer != null) {
kafkaProducer.close();
}
}
@Override
public void processElement(Long timestamp, Context ctx,
Collector<Tuple2<Long, Integer>> out) throws Exception {
//timer coalescing avoids firing more than once per timestamp
ctx.timerService().registerEventTimeTimer(timestamp);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<Tuple2<Long, Integer>> out) throws Exception {
//Flink guarantees operator will only fire after
normal sink has finished producing messages
//send one event per partition downstream
kafkaProducer.partitionsFor("OUTPUT_TOPIC").forEach(partitionInfo ->
out.collect(new Tuple2<>(timestamp, partitionInfo.partition())));
}
}).addSink(
new FlinkKafkaProducer<Tuple2<Long,
Integer>>(KAFKA_SINK_TOPIC, serializationSchema, producerProperties,
Optional.of(new FlinkKafkaPartitioner<Tuple2<Long, Integer>>() {
@Override
public int partition(Tuple2<Long, Integer> record, byte[]
key, byte[] value, String targetTopic, int[] partitions) {
return record._2(); //send to the partition it was
designed to be sent
}
}), FlinkKafkaProducer.Semantic.EXACTLY_ONCE, kafkaProducerPoolSize));
On Mon, Sep 2, 2019 at 5:07 PM Padarn Wilson <pa...@gmail.com> wrote:
>
> Hi Fabian,
>
> > but each partition may only be written by a single task
>
> Sorry I think I misunderstand something here then: If I have a topic with one partition, but multiple sink tasks (or parallelism > 1).. this means the data must all be shuffled to the single task writing that partition?
>
> Padarn
>
> On Mon, Sep 2, 2019 at 9:52 PM Fabian Hueske <fh...@gmail.com> wrote:
>>
>> Hi Padarn,
>>
>> Regarding your throughput concerns: A sink task may write to multiple partitions, but each partition may only be written by a single task.
>>
>> @Eduardo: Thanks for sharing your approach! Not sure if I understood it correctly, but I think that the approach does not guarantee that all results of a window are emitted before the end-of-window marker is written.
>> Since the sink operator and the single-task-operator are separate operators, the output records might get stuck (or be bufffered) in one of the sink tasks and the single-task would still emit an end-of-window marker record because it doesn't know about the sink task.
>>
>> Best,
>> Fabian
>>
>> Am Do., 29. Aug. 2019 um 18:42 Uhr schrieb Eduardo Winpenny Tejedor <ed...@gmail.com>:
>>>
>>> Hi,
>>>
>>> I'll chip in with an approach I'm trying at the moment that seems to work, and I say seems because I'm only running this on a personal project.
>>>
>>> Personally, I don't have anything against end-of-message markers per partition, Padarn you seem to not prefer this option as it overloads the meaning of the output payload. My approach is equally valid when producing watermarks/end-of-message markers on a side output though.
>>>
>>> The main problem of both approaches is knowing when the window has finished across all partitions without having to wait for the start of the next window.
>>>
>>> I've taken the approach of sending all output messages of the window to 1. the sink but also 2. a single task operator. The single task operator registers an event time based timer at the time of the end of the window. You have the confidence of the task's timer triggering only once at the right time because all the post-window watermarks go through to the same task. At that point I make the task send an end-of-message marker to every partition. I don't need to send the count because Kafka messages are ordered. AND IF you prefer to not overload the semantic of your original Kafka topic you can post the message to a separate location of your choice.
>>>
>>> While this does mean that the end of marker message only gets sent through once the window has finished across all substreams (as opposed to per stream), it does mean you don't need to wait for the next window to start AND the watermark gap between substreams should never grow that much anyway.
>>>
>>> This approach should be particularly useful when the number of partitions or keying mechanism is different between the input and output topics.
>>>
>>> Hopefully that doesn't sound like a terrible idea.
>>>
>>> eduardo
>>>
>>>
>>>
>>>
>>> On Wed, 28 Aug 2019, 02:54 Padarn Wilson, <pa...@gmail.com> wrote:
>>>>
>>>> Hi again Fabian,
>>>>
>>>> Thanks for pointing this out to me. In my case there is no need for keyed writing - but I do wonder if having each kafka task write only to a single partition would significantly affect performance.
>>>>
>>>> Actually now that I think about it, the approach to just wait for the first records of the next window is also subject to the problem you mention above: a producer lagging behind the rest could end up with a partition containing element out of ‘window order’.
>>>>
>>>> I was also thinking this problem is very similar to that of checkpoint barriers. I intended to dig into the details of the exactly once Kafka sink for some inspiration.
>>>>
>>>> Padarn
>>>>
>>>> On Tue, 27 Aug 2019 at 11:01 PM, Fabian Hueske <fh...@gmail.com> wrote:
>>>>>
>>>>> Hi Padarn,
>>>>>
>>>>> Yes, this is quite tricky.
>>>>> The "problem" with watermarks is that you need to consider how you write to Kafka.
>>>>> If your Kafka sink writes to keyed Kafka stream (each Kafka partition is written by multiple producers), you need to broadcast the watermarks to each partition, i.e., each partition would receive watermarks from each parallel sink task. So in order to reason about the current watermark of a partition, you need to observe them and take the minimum WM across all current sink task WMs.
>>>>> Things become much easier, if each partition is only written by a single task but this also means that data is not key-partitioned in Kafka.
>>>>> In that case, the sink task only needs to write a WM message to each of its assigned partitions.
>>>>>
>>>>> Hope this helps,
>>>>> Fabian
>>>>>
>>>>>
>>>>> Am Sa., 17. Aug. 2019 um 05:48 Uhr schrieb Padarn Wilson <pa...@gmail.com>:
>>>>>>
>>>>>> Hi Fabian, thanks for your input
>>>>>>
>>>>>> Exactly. Actually my first instinct was to see if it was possible to publish the watermarks somehow - my initial idea was to insert regular watermark messages into each partition of the stream, but exposing this seemed quite troublesome.
>>>>>>
>>>>>> > In that case, you could have a ProcessFunction that is chained before the sink and which counts the window results per time slice and emits the result when the watermark passes to a side output.
>>>>>> All side output messages are collected by a single task and can be published to a Kafka topic or even be made available via Queryable State.
>>>>>>
>>>>>> I understand the idea here (and exactly once semantics are probably fine for my use case), but counting events seems a bit fragile. I'm not totally confident the consumer can guarantee it won't read duplicates (its a golang kafka library that seems to have some quirks).
>>>>>>
>>>>>> I think ideally each partition of the kafka topic would have some regular information about watermarks. Perhaps the kafka producer can be modified to support this.
>>>>>>
>>>>>> Padarn
>>>>>>
>>>>>> On Fri, Aug 16, 2019 at 3:50 PM Fabian Hueske <fh...@gmail.com> wrote:
>>>>>>>
>>>>>>> Hi Padarn,
>>>>>>>
>>>>>>> What you describe is essentially publishing Flink's watermarks to an outside system.
>>>>>>> Flink processes time windows, by waiting for a watermark that's past the window end time. When it receives such a WM it processes and emits all ended windows and forwards the watermark.
>>>>>>> When a sink received a WM for say 12:45:15, you know that all window results with until 12:45:00 have been emitted.
>>>>>>> Hence, the watermark tells you about the completeness of data.
>>>>>>>
>>>>>>> However, using this information is not so easy, mostly because of the failure semantics.
>>>>>>> Things become much easier if you produce to Kafka with exactly-once semantics.
>>>>>>>
>>>>>>> In that case, you could have a ProcessFunction that is chained before the sink and which counts the window results per time slice and emits the result when the watermark passes to a side output.
>>>>>>> All side output messages are collected by a single task and can be published to a Kafka topic or even be made available via Queryable State.
>>>>>>>
>>>>>>> For at-least once output, it's much harder because you'll have duplicates in the output after a job recovered.
>>>>>>>
>>>>>>> Best, Fabian
>>>>>>>
>>>>>>> I think you have two options to let the consuming app know about the progress.
>>>>>>> You can either
>>>>>>>
>>>>>>> The ProcessFunction could count per window end timestamp how many records passed and forward that information via a side output.
>>>>>>> You could then
>>>>>>>
>>>>>>>
>>>>>>> Essentially, you'd like to publish Flink's watermarks to an outside system (possibly via Kafka).
>>>>>>>
>>>>>>>
>>>>>>> Am Mo., 12. Aug. 2019 um 14:33 Uhr schrieb Padarn Wilson <pa...@gmail.com>:
>>>>>>>>
>>>>>>>> Hello Users,
>>>>>>>>
>>>>>>>> I have a question that is perhaps not best solved within Flink: It has to do with notifying a downstream application that a Flink window has completed.
>>>>>>>>
>>>>>>>> The (simplified) scenario is this:
>>>>>>>> - We have a Flink job that consumes from Kafka, does some preprocessing, and then has a sliding window of 10 minutes and slide time of 1 minute.
>>>>>>>> - The number of keys in each slide is not fixed
>>>>>>>> - The output of the window is then output to Kafka, which is read by a downstream application.
>>>>>>>>
>>>>>>>> What I want to achieve is that the downstream application can someone know when it has read all of the data for a single window, without waiting for the next window to arrive.
>>>>>>>>
>>>>>>>> Some options I've considered:
>>>>>>>> - Producing a second window over the window results that counts the output size, which can then be used by the downstream application to see when it has received the same number: This seems fragile, as there it relies on there being no loss or duplication of data. Its also an extra window and Kafka stream which is a tad messy.
>>>>>>>> - Somehow adding an 'end of window' element to each partitions of the Kafka topic which can be read by the consumer: This seems a bit messy because it mixes different types of events into the same Kafka stream, and there is no really simple way to do this in Flink
>>>>>>>> - Package the whole window output into a single message and make this the unit of transaction: This is possible, but the message would be quite large then (at least 10s of mb), as the volume of this stream is quite large.
>>>>>>>> - Assume that if the consumer has no elements to read, or if the next window has started to be read, then it has read the whole window: This seems reasonable, and if it wasn't for the fact that my consumer on the application end was a bit inflexible right now, it is probably the solution I would use.
>>>>>>>>
>>>>>>>> Any further/better ideas?
>>>>>>>>
>>>>>>>> Thanks
>>>>>>>> Padarn
Re: End of Window Marker
Posted by Padarn Wilson <pa...@gmail.com>.
Hi Fabian,
> but each partition may only be written by a single task
Sorry I think I misunderstand something here then: If I have a topic with
one partition, but multiple sink tasks (or parallelism > 1).. this means
the data must all be shuffled to the single task writing that partition?
Padarn
On Mon, Sep 2, 2019 at 9:52 PM Fabian Hueske <fh...@gmail.com> wrote:
> Hi Padarn,
>
> Regarding your throughput concerns: A sink task may write to multiple
> partitions, but each partition may only be written by a single task.
>
> @Eduardo: Thanks for sharing your approach! Not sure if I understood it
> correctly, but I think that the approach does not guarantee that all
> results of a window are emitted before the end-of-window marker is written.
> Since the sink operator and the single-task-operator are separate
> operators, the output records might get stuck (or be bufffered) in one of
> the sink tasks and the single-task would still emit an end-of-window marker
> record because it doesn't know about the sink task.
>
> Best,
> Fabian
>
> Am Do., 29. Aug. 2019 um 18:42 Uhr schrieb Eduardo Winpenny Tejedor <
> eduardo.winpenny@gmail.com>:
>
>> Hi,
>>
>> I'll chip in with an approach I'm trying at the moment that seems to
>> work, and I say seems because I'm only running this on a personal project.
>>
>> Personally, I don't have anything against end-of-message markers per
>> partition, Padarn you seem to not prefer this option as it overloads the
>> meaning of the output payload. My approach is equally valid when producing
>> watermarks/end-of-message markers on a side output though.
>>
>> The main problem of both approaches is knowing when the window has
>> finished across all partitions without having to wait for the start of the
>> next window.
>>
>> I've taken the approach of sending all output messages of the window to
>> 1. the sink but also 2. a single task operator. The single task operator
>> registers an event time based timer at the time of the end of the window.
>> You have the confidence of the task's timer triggering only once at the
>> right time because all the post-window watermarks go through to the same
>> task. At that point I make the task send an end-of-message marker to every
>> partition. I don't need to send the count because Kafka messages are
>> ordered. AND IF you prefer to not overload the semantic of your original
>> Kafka topic you can post the message to a separate location of your choice.
>>
>> While this does mean that the end of marker message only gets sent
>> through once the window has finished across all substreams (as opposed to
>> per stream), it does mean you don't need to wait for the next window to
>> start AND the watermark gap between substreams should never grow that much
>> anyway.
>>
>> This approach should be particularly useful when the number of partitions
>> or keying mechanism is different between the input and output topics.
>>
>> Hopefully that doesn't sound like a terrible idea.
>>
>> eduardo
>>
>>
>>
>>
>> On Wed, 28 Aug 2019, 02:54 Padarn Wilson, <pa...@gmail.com> wrote:
>>
>>> Hi again Fabian,
>>>
>>> Thanks for pointing this out to me. In my case there is no need for
>>> keyed writing - but I do wonder if having each kafka task write only to a
>>> single partition would significantly affect performance.
>>>
>>> Actually now that I think about it, the approach to just wait for the
>>> first records of the next window is also subject to the problem you mention
>>> above: a producer lagging behind the rest could end up with a partition
>>> containing element out of ‘window order’.
>>>
>>> I was also thinking this problem is very similar to that of checkpoint
>>> barriers. I intended to dig into the details of the exactly once Kafka sink
>>> for some inspiration.
>>>
>>> Padarn
>>>
>>> On Tue, 27 Aug 2019 at 11:01 PM, Fabian Hueske <fh...@gmail.com>
>>> wrote:
>>>
>>>> Hi Padarn,
>>>>
>>>> Yes, this is quite tricky.
>>>> The "problem" with watermarks is that you need to consider how you
>>>> write to Kafka.
>>>> If your Kafka sink writes to keyed Kafka stream (each Kafka partition
>>>> is written by multiple producers), you need to broadcast the watermarks to
>>>> each partition, i.e., each partition would receive watermarks from each
>>>> parallel sink task. So in order to reason about the current watermark of a
>>>> partition, you need to observe them and take the minimum WM across all
>>>> current sink task WMs.
>>>> Things become much easier, if each partition is only written by a
>>>> single task but this also means that data is not key-partitioned in Kafka.
>>>> In that case, the sink task only needs to write a WM message to each of
>>>> its assigned partitions.
>>>>
>>>> Hope this helps,
>>>> Fabian
>>>>
>>>>
>>>> Am Sa., 17. Aug. 2019 um 05:48 Uhr schrieb Padarn Wilson <
>>>> padarn@gmail.com>:
>>>>
>>>>> Hi Fabian, thanks for your input
>>>>>
>>>>> Exactly. Actually my first instinct was to see if it was possible to
>>>>> publish the watermarks somehow - my initial idea was to insert regular
>>>>> watermark messages into each partition of the stream, but exposing this
>>>>> seemed quite troublesome.
>>>>>
>>>>> > In that case, you could have a ProcessFunction that is chained
>>>>> before the sink and which counts the window results per time slice and
>>>>> emits the result when the watermark passes to a side output.
>>>>> All side output messages are collected by a single task and can be
>>>>> published to a Kafka topic or even be made available via Queryable State.
>>>>>
>>>>> I understand the idea here (and exactly once semantics are probably
>>>>> fine for my use case), but counting events seems a bit fragile. I'm not
>>>>> totally confident the consumer can guarantee it won't read duplicates (its
>>>>> a golang kafka library that seems to have some quirks).
>>>>>
>>>>> I think ideally each partition of the kafka topic would have some
>>>>> regular information about watermarks. Perhaps the kafka producer can be
>>>>> modified to support this.
>>>>>
>>>>> Padarn
>>>>>
>>>>> On Fri, Aug 16, 2019 at 3:50 PM Fabian Hueske <fh...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Hi Padarn,
>>>>>>
>>>>>> What you describe is essentially publishing Flink's watermarks to an
>>>>>> outside system.
>>>>>> Flink processes time windows, by waiting for a watermark that's past
>>>>>> the window end time. When it receives such a WM it processes and emits all
>>>>>> ended windows and forwards the watermark.
>>>>>> When a sink received a WM for say 12:45:15, you know that all window
>>>>>> results with until 12:45:00 have been emitted.
>>>>>> Hence, the watermark tells you about the completeness of data.
>>>>>>
>>>>>> However, using this information is not so easy, mostly because of the
>>>>>> failure semantics.
>>>>>> Things become much easier if you produce to Kafka with exactly-once
>>>>>> semantics.
>>>>>>
>>>>>> In that case, you could have a ProcessFunction that is chained before
>>>>>> the sink and which counts the window results per time slice and emits the
>>>>>> result when the watermark passes to a side output.
>>>>>> All side output messages are collected by a single task and can be
>>>>>> published to a Kafka topic or even be made available via Queryable State.
>>>>>>
>>>>>> For at-least once output, it's much harder because you'll have
>>>>>> duplicates in the output after a job recovered.
>>>>>>
>>>>>> Best, Fabian
>>>>>>
>>>>>> I think you have two options to let the consuming app know about the
>>>>>> progress.
>>>>>> You can either
>>>>>>
>>>>>> The ProcessFunction could count per window end timestamp how many
>>>>>> records passed and forward that information via a side output.
>>>>>> You could then
>>>>>>
>>>>>>
>>>>>> Essentially, you'd like to publish Flink's watermarks to an outside
>>>>>> system (possibly via Kafka).
>>>>>>
>>>>>>
>>>>>> Am Mo., 12. Aug. 2019 um 14:33 Uhr schrieb Padarn Wilson <
>>>>>> padarn@gmail.com>:
>>>>>>
>>>>>>> Hello Users,
>>>>>>>
>>>>>>> I have a question that is perhaps not best solved within Flink: It
>>>>>>> has to do with notifying a downstream application that a Flink window has
>>>>>>> completed.
>>>>>>>
>>>>>>> The (simplified) scenario is this:
>>>>>>> - We have a Flink job that consumes from Kafka, does some
>>>>>>> preprocessing, and then has a sliding window of 10 minutes and slide time
>>>>>>> of 1 minute.
>>>>>>> - The number of keys in each slide is not fixed
>>>>>>> - The output of the window is then output to Kafka, which is read by
>>>>>>> a downstream application.
>>>>>>>
>>>>>>> What I want to achieve is that the downstream application can
>>>>>>> someone know when it has read all of the data for a single window, without
>>>>>>> waiting for the next window to arrive.
>>>>>>>
>>>>>>> Some options I've considered:
>>>>>>> - Producing a second window over the window results that counts the
>>>>>>> output size, which can then be used by the downstream application to see
>>>>>>> when it has received the same number: This seems fragile, as there it
>>>>>>> relies on there being no loss or duplication of data. Its also an extra
>>>>>>> window and Kafka stream which is a tad messy.
>>>>>>> - Somehow adding an 'end of window' element to each partitions of
>>>>>>> the Kafka topic which can be read by the consumer: This seems a bit messy
>>>>>>> because it mixes different types of events into the same Kafka stream, and
>>>>>>> there is no really simple way to do this in Flink
>>>>>>> - Package the whole window output into a single message and make
>>>>>>> this the unit of transaction: This is possible, but the message would be
>>>>>>> quite large then (at least 10s of mb), as the volume of this stream is
>>>>>>> quite large.
>>>>>>> - Assume that if the consumer has no elements to read, or if the
>>>>>>> next window has started to be read, then it has read the whole window: This
>>>>>>> seems reasonable, and if it wasn't for the fact that my consumer on the
>>>>>>> application end was a bit inflexible right now, it is probably the solution
>>>>>>> I would use.
>>>>>>>
>>>>>>> Any further/better ideas?
>>>>>>>
>>>>>>> Thanks
>>>>>>> Padarn
>>>>>>>
>>>>>>