You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Juan Gentile <j....@criteo.com> on 2019/07/03 16:04:21 UTC
Watermarks and Kafka
Hello!
We currently have a job which reads from Kafka and uses punctuated watermarks based on the messages we read. We currently keep track of the watermarks for each partition to emit a consensus watermark, taking the smallest of all partitions.
We ran into an issue because we are not storing the state of this map of partitions->watermarks when one of the partitions got delayed and the job restarted, losing track of that partition and emitting a watermark anyway.
Our idea of a solution involves saving this map of partition -> watermarks into the state but we would like to know how Flink behaves when we decrease the parallelism so as to make sure that the instance that will read from Kafka also will have the state for that particular partition.
To give an example:
Operator 1: (Reads Partition1)
Partition 1: Watermark1 (Map / State)
Operator 2: (Reads Partition2)
Partition 2: Watermark2 (Map / State)
Operator 3: (Reads Partition1)
Partition 3: Watermark3 (Map / State)
After shrinking:
Operator 1: (Reads Partition1)
Partition 1: Watermark1 (Map / State)
Operator 2: (Reads Partition2, Partition3)
Partition 2: Watermark2 (Map / State)
Partition 3: Watermark3 (Map / State)
Or
Operator 1: (Reads Partition1, Partition3) => HERE we would have a problem as the state could be loaded on the other operator.
Partition 1: Watermark1 (Map / State)
Operator 2: (Reads Partition2)
Partition 2: Watermark2 (Map / State)
Partition 3: Watermark3 (Map / State)
For this we are using the operator state (https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-managed-operator-state) with “Even-split redistribution”
Could you please give us a hand understanding how Flink behaves in such scenario?
Thank you,
Juan G.
Re: Watermarks and Kafka
Posted by Juan Gentile <j....@criteo.com>.
Hello Konstantin,
Thank you for you answer, I’ll clarify a bit our problem as actually we have a clear understanding of our problem now 😊.
We have 2 Kafka topics from 2 different datacenters (each with its own watermarks – We have a watermark message injected in each of them).
We replicate these into a single Kafka topic from which our Flink job consumes. In the Flink job, we consume per partition but the watermarks in a partition may come from 2 different datacenters, so the watermark could differ. That’s why we need to keep a Map in memory for each partition and DC and then emit the minimum of both.
Our current workaround to the problem is to implement something similar to what you mentioned, we have an operator that assigns the watermarks and has a Map which is stored in the state. But since there is no guarantee that all of these operators will receive messages from all partitions we need to remove the partition from the Map, and just use DC’s. Then we use the union redistribution and always get the minimum of all DC’s for all (when it restores). This seems to work if we keep the same parallelism for the source and the watermark assigner, keeping them as close as possible in the flow and use operator chaining. But we understand that if we were to split them or have different parallelism then the watermark assigner would stop working properly because the partition wouldn’t be in the map. So that’s why we were looking for a solution that has already the watermarks handled in the source operator.
Please let us know your opinion.
Thank you,
Juan G.
From: Konstantin Knauf <ko...@ververica.com>
Date: Sunday, July 7, 2019 at 10:14 PM
To: Juan Gentile <j....@criteo.com>
Cc: "user@flink.apache.org" <us...@flink.apache.org>, Olivier Solliec <o....@criteo.com>, Oleksandr Nitavskyi <o....@criteo.com>
Subject: Re: Watermarks and Kafka
Hi Juan,
I just replied to your other question, but I think, I better get where you are coming from now.
Are you aware of per-partition watermarking [1]? You don't need to manage this map yourself. BUT: this does not solve the problem, that this Map is not stored in Managed State. Watermarks are generally not part of Flink's State. It seems like this is what you are looking for?
To also answer your questions: You could go for List<Entry<Partition, Watermark>> state with union redistribution. In this case every operator will get all entries during recovery and you can filter out the ones, which are relevant to the current operator by checking which partitions it is subscribed to after recovery.
Hope this helps.
Cheers,
Konstantin
[1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/connectors/kafka.html#kafka-consumers-and-timestamp-extractionwatermark-emission<https://eur03.safelinks.protection.outlook.com/?url=https%3A%2F%2Fci.apache.org%2Fprojects%2Fflink%2Fflink-docs-release-1.8%2Fdev%2Fconnectors%2Fkafka.html%23kafka-consumers-and-timestamp-extractionwatermark-emission&data=02%7C01%7Cj.gentile%40criteo.com%7C6142c5cb95b34bb356ee08d70317a4c3%7C2a35d8fd574d48e3927c8c398e225a01%7C1%7C0%7C636981272400501401&sdata=auakXJ9kguK8%2B3etpTHK9f5gW0V1TcWxAkGP4RIiSCw%3D&reserved=0>
On Wed, Jul 3, 2019 at 6:04 PM Juan Gentile <j....@criteo.com>> wrote:
Hello!
We currently have a job which reads from Kafka and uses punctuated watermarks based on the messages we read. We currently keep track of the watermarks for each partition to emit a consensus watermark, taking the smallest of all partitions.
We ran into an issue because we are not storing the state of this map of partitions->watermarks when one of the partitions got delayed and the job restarted, losing track of that partition and emitting a watermark anyway.
Our idea of a solution involves saving this map of partition -> watermarks into the state but we would like to know how Flink behaves when we decrease the parallelism so as to make sure that the instance that will read from Kafka also will have the state for that particular partition.
To give an example:
Operator 1: (Reads Partition1)
Partition 1: Watermark1 (Map / State)
Operator 2: (Reads Partition2)
Partition 2: Watermark2 (Map / State)
Operator 3: (Reads Partition1)
Partition 3: Watermark3 (Map / State)
After shrinking:
Operator 1: (Reads Partition1)
Partition 1: Watermark1 (Map / State)
Operator 2: (Reads Partition2, Partition3)
Partition 2: Watermark2 (Map / State)
Partition 3: Watermark3 (Map / State)
Or
Operator 1: (Reads Partition1, Partition3) => HERE we would have a problem as the state could be loaded on the other operator.
Partition 1: Watermark1 (Map / State)
Operator 2: (Reads Partition2)
Partition 2: Watermark2 (Map / State)
Partition 3: Watermark3 (Map / State)
For this we are using the operator state (https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-managed-operator-state<https://eur03.safelinks.protection.outlook.com/?url=https%3A%2F%2Fci.apache.org%2Fprojects%2Fflink%2Fflink-docs-stable%2Fdev%2Fstream%2Fstate%2Fstate.html%23using-managed-operator-state&data=02%7C01%7Cj.gentile%40criteo.com%7C6142c5cb95b34bb356ee08d70317a4c3%7C2a35d8fd574d48e3927c8c398e225a01%7C1%7C0%7C636981272400501401&sdata=x0u1aFZbywIzT13%2F3v8TlnNWmJmc%2Bhkp2nh87X%2B9fnM%3D&reserved=0>) with “Even-split redistribution”
Could you please give us a hand understanding how Flink behaves in such scenario?
Thank you,
Juan G.
--
Konstantin Knauf | Solutions Architect
+49 160 91394525
Planned Absences: 10.08.2019 - 31.08.2019, 05.09. - 06.09.2010
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Dr. Kostas Tzoumas, Dr. Stephan Ewen
Re: Watermarks and Kafka
Posted by Konstantin Knauf <ko...@ververica.com>.
Hi Juan,
I just replied to your other question, but I think, I better get where you
are coming from now.
Are you aware of per-partition watermarking [1]? You don't need to manage
this map yourself. BUT: this does not solve the problem, that this Map is
not stored in Managed State. Watermarks are generally not part of Flink's
State. It seems like this is what you are looking for?
To also answer your questions: You could go for List<Entry<Partition,
Watermark>> state with union redistribution. In this case every operator
will get all entries during recovery and you can filter out the ones, which
are relevant to the current operator by checking which partitions it is
subscribed to after recovery.
Hope this helps.
Cheers,
Konstantin
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/connectors/kafka.html#kafka-consumers-and-timestamp-extractionwatermark-emission
On Wed, Jul 3, 2019 at 6:04 PM Juan Gentile <j....@criteo.com> wrote:
> Hello!
>
>
>
> We currently have a job which reads from Kafka and uses punctuated
> watermarks based on the messages we read. We currently keep track of the
> watermarks for each partition to emit a consensus watermark, taking the
> smallest of all partitions.
>
> We ran into an issue because we are not storing the state of this map of
> partitions->watermarks when one of the partitions got delayed and the job
> restarted, losing track of that partition and emitting a watermark anyway.
>
> Our idea of a solution involves saving this map of partition -> watermarks
> into the state but we would like to know how Flink behaves when we decrease
> the parallelism so as to make sure that the instance that will read from
> Kafka also will have the state for that particular partition.
>
>
>
> To give an example:
>
>
>
> Operator 1: (Reads Partition1)
>
> Partition 1: Watermark1 (Map / State)
>
>
>
> Operator 2: (Reads Partition2)
>
> Partition 2: Watermark2 (Map / State)
>
>
>
> Operator 3: (Reads Partition1)
>
> Partition 3: Watermark3 (Map / State)
>
>
>
>
>
> After shrinking:
>
>
>
> Operator 1: (Reads Partition1)
>
> Partition 1: Watermark1 (Map / State)
>
>
>
> Operator 2: (Reads Partition2, Partition3)
>
> Partition 2: Watermark2 (Map / State)
>
> Partition 3: Watermark3 (Map / State)
>
>
>
> Or
>
>
>
> Operator 1: (Reads Partition1, Partition3) => HERE we would have a
> problem as the state could be loaded on the other operator.
>
> Partition 1: Watermark1 (Map / State)
>
>
>
> Operator 2: (Reads Partition2)
>
> Partition 2: Watermark2 (Map / State)
>
> Partition 3: Watermark3 (Map / State)
>
>
>
> For this we are using the operator state (
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-managed-operator-state)
> with “*Even-split redistribution*”
>
>
>
> Could you please give us a hand understanding how Flink behaves in such
> scenario?
>
>
>
> Thank you,
>
> Juan G.
>
--
Konstantin Knauf | Solutions Architect
+49 160 91394525
Planned Absences: 10.08.2019 - 31.08.2019, 05.09. - 06.09.2010
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Dr. Kostas Tzoumas, Dr. Stephan Ewen