You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by "Kathula, Sandeep" <Sa...@intuit.com> on 2021/11/15 17:22:22 UTC
Beam on Flink runner not able to advance watermarks on a high load
Hi,
We are running a Beam application on Flink runner (Beam 2.29 and Flink 1.12) which reads from Kafka and writes to S3 once every 5 minutes. My window and s3 writes looks like
PCollection<GenericRecord>.apply("Batch Events", Window.<GenericRecord>into(
FixedWindows.of(Duration.standardMinutes(5)))
.triggering(AfterWatermark.pastEndOfWindow())
.withAllowedLateness(Duration.ZERO, Window.ClosingBehavior.FIRE_ALWAYS)
.discardingFiredPanes())
.apply(FileIO.<GenericRecord>write()
.via(ParquetIO.sink(schema))
.to(outputPath)
.withNumShards(5)
.withNaming(new CustomFileNaming("snappy.parquet")));
Resources allocated: 5 task slots each with 3 CPU and 32 GB RAM. We are using RocksDB as state backend and giving 50% of memory to off-heap.
Its running fine with lighter loads. But when it gets heavier load from Kafka (7500 or more records per sec – each record around 7KB in size), we are seeing that no files are being written to S3.We are using AfterWatermark.pastEndOfWindow() which is trigerring only when the watermark reaches the end of window.
After debugging we found that watermarks are not being advanced during heavy loads and as a result event time triggers after watermark reaches end of window because of which s3 writes will happen are not getting triggered. So the data is accumulating in off-heap which results in out of memory after some time.
Can you please let us know why watermarks are not advancing under high load.
Thanks,
Sandeep
Re: Beam on Flink runner not able to advance watermarks on a high load
Posted by Jan Lukavský <je...@seznam.cz>.
Hi Sandeep,
if you are certain, that the timestamps of the messages in Kafka topic
you are processing are correctly generated (which probably should be, if
adding more resources 'fixes' the issue), I suggest you try to create a
flame graph, or any other form of aggregated view over hotspots in
thread dump. You might want to as well verify, that your pipeline
processes records from all partitions of your topic(s), so that no
partition is accidentally left-behind. But that is all that comes to my
mind, currently. If you'll have any more insights, do not hesitate to
share, so that the community can help figure this out, because that
behavior 'feels' unexpected - at least to me.
Best,
Jan
On 11/29/21 19:45, Kathula, Sandeep wrote:
>
> Hi Jan,
>
> We are already using --experiments=use_deprecated_read and
> Flink is not advancing the watermarks.
>
> Thanks,
>
> Sandeep
>
> *From: *Jan Lukavský <je...@seznam.cz>
> *Date: *Thursday, November 18, 2021 at 4:03 AM
> *To: *"Kathula, Sandeep" <Sa...@intuit.com>,
> "user@beam.apache.org" <us...@beam.apache.org>
> *Subject: *Re: Beam on Flink runner not able to advance watermarks on
> a high load
>
> This email is from an external sender.
>
> Hi Sandeep,
>
> one more question, did you try to use
> --experiments=use_deprecated_read? If not, can you try that and check
> if it has any impact on the behavior you observe?
>
> Jan
>
> On 11/18/21 01:41, Kathula, Sandeep wrote:
>
> Hi Jan,
>
> We are not adding any custom timestamp policy. We also don’t see
> backpressure in Flink UI. We are giving 5 task slots each with 3
> CPU and 32 GB RAM. Its working if we give 10 task slots each with
> 3 CPU and 32 GB RAM. But that’s lot of resources for this load. We
> are trying to figure out why Beam is not able to handle 10,000
> records per second with 5 task slots.
>
> Thanks,
>
> Sandeep
>
> *From: *Jan Lukavský <je...@seznam.cz> <ma...@seznam.cz>
> *Reply-To: *"user@beam.apache.org" <ma...@beam.apache.org>
> <us...@beam.apache.org> <ma...@beam.apache.org>
> *Date: *Tuesday, November 16, 2021 at 3:11 AM
> *To: *"user@beam.apache.org" <ma...@beam.apache.org>
> <us...@beam.apache.org> <ma...@beam.apache.org>
> *Subject: *Re: Beam on Flink runner not able to advance watermarks
> on a high load
>
> This email is from an external sender.
>
> Hi Sandeep,
>
> - dev@beam <ma...@beam.apache.org>
>
> The watermark estimation itself should not be related to load. Can
> you please clarify, if
>
> a) you are using any custom timestamp policy?
>
> b) you see any backpressure in Flink's UI? Backpressure could -
> under some circumstances - cause delays in watermark propagation.
> It _might_ help to increase parallelism in that case.
>
> Best,
>
> Jan
>
> On 11/15/21 18:22, Kathula, Sandeep wrote:
>
> Hi,
>
> We are running a Beam application on Flink runner (Beam 2.29
> and Flink 1.12) which reads from Kafka and writes to S3 once
> every 5 minutes. My window and s3 writes looks like
>
> PCollection<GenericRecord>.apply("Batch Events",
> Window.<GenericRecord>into(
>
> FixedWindows.of(Duration.standardMinutes(5)))
>
> .triggering(AfterWatermark.pastEndOfWindow())
>
> .withAllowedLateness(Duration.ZERO,
> Window.ClosingBehavior.FIRE_ALWAYS)
>
> .discardingFiredPanes())
>
> .apply(FileIO.<GenericRecord>write()
>
> .via(ParquetIO.sink(schema))
>
> .to(outputPath)
>
> .withNumShards(5)
>
> .withNaming(new CustomFileNaming("snappy.parquet")));
>
> Resources allocated: 5 task slots each with 3 CPU and 32 GB
> RAM. We are using RocksDB as state backend and giving 50% of
> memory to off-heap.
>
> Its running fine with lighter loads. But when it gets heavier
> load from Kafka (7500 or more records per sec – each record
> around 7KB in size), we are seeing that no files are being
> written to S3.We are using AfterWatermark.pastEndOfWindow()
> which is trigerring only when the watermark reaches the end of
> window.
>
> After debugging we found that watermarks are not being
> advanced during heavy loads and as a result event time
> triggers after watermark reaches end of window because of
> which s3 writes will happen are not getting triggered. So the
> data is accumulating in off-heap which results in out of
> memory after some time.
>
> Can you please let us know why watermarks are not advancing
> under high load.
>
> Thanks,
>
> Sandeep
>
Re: Beam on Flink runner not able to advance watermarks on a high load
Posted by "Kathula, Sandeep" <Sa...@intuit.com>.
Hi Jan,
We are already using --experiments=use_deprecated_read and Flink is not advancing the watermarks.
Thanks,
Sandeep
From: Jan Lukavský <je...@seznam.cz>
Date: Thursday, November 18, 2021 at 4:03 AM
To: "Kathula, Sandeep" <Sa...@intuit.com>, "user@beam.apache.org" <us...@beam.apache.org>
Subject: Re: Beam on Flink runner not able to advance watermarks on a high load
This email is from an external sender.
Hi Sandeep,
one more question, did you try to use --experiments=use_deprecated_read? If not, can you try that and check if it has any impact on the behavior you observe?
Jan
On 11/18/21 01:41, Kathula, Sandeep wrote:
Hi Jan,
We are not adding any custom timestamp policy. We also don’t see backpressure in Flink UI. We are giving 5 task slots each with 3 CPU and 32 GB RAM. Its working if we give 10 task slots each with 3 CPU and 32 GB RAM. But that’s lot of resources for this load. We are trying to figure out why Beam is not able to handle 10,000 records per second with 5 task slots.
Thanks,
Sandeep
From: Jan Lukavský <je...@seznam.cz>
Reply-To: "user@beam.apache.org"<ma...@beam.apache.org> <us...@beam.apache.org>
Date: Tuesday, November 16, 2021 at 3:11 AM
To: "user@beam.apache.org"<ma...@beam.apache.org> <us...@beam.apache.org>
Subject: Re: Beam on Flink runner not able to advance watermarks on a high load
This email is from an external sender.
Hi Sandeep,
- dev@beam<ma...@beam.apache.org>
The watermark estimation itself should not be related to load. Can you please clarify, if
a) you are using any custom timestamp policy?
b) you see any backpressure in Flink's UI? Backpressure could - under some circumstances - cause delays in watermark propagation. It _might_ help to increase parallelism in that case.
Best,
Jan
On 11/15/21 18:22, Kathula, Sandeep wrote:
Hi,
We are running a Beam application on Flink runner (Beam 2.29 and Flink 1.12) which reads from Kafka and writes to S3 once every 5 minutes. My window and s3 writes looks like
PCollection<GenericRecord>.apply("Batch Events", Window.<GenericRecord>into(
FixedWindows.of(Duration.standardMinutes(5)))
.triggering(AfterWatermark.pastEndOfWindow())
.withAllowedLateness(Duration.ZERO, Window.ClosingBehavior.FIRE_ALWAYS)
.discardingFiredPanes())
.apply(FileIO.<GenericRecord>write()
.via(ParquetIO.sink(schema))
.to(outputPath)
.withNumShards(5)
.withNaming(new CustomFileNaming("snappy.parquet")));
Resources allocated: 5 task slots each with 3 CPU and 32 GB RAM. We are using RocksDB as state backend and giving 50% of memory to off-heap.
Its running fine with lighter loads. But when it gets heavier load from Kafka (7500 or more records per sec – each record around 7KB in size), we are seeing that no files are being written to S3.We are using AfterWatermark.pastEndOfWindow() which is trigerring only when the watermark reaches the end of window.
After debugging we found that watermarks are not being advanced during heavy loads and as a result event time triggers after watermark reaches end of window because of which s3 writes will happen are not getting triggered. So the data is accumulating in off-heap which results in out of memory after some time.
Can you please let us know why watermarks are not advancing under high load.
Thanks,
Sandeep
Re: Beam on Flink runner not able to advance watermarks on a high load
Posted by Jan Lukavský <je...@seznam.cz>.
Hi Sandeep,
one more question, did you try to use --experiments=use_deprecated_read?
If not, can you try that and check if it has any impact on the behavior
you observe?
Jan
On 11/18/21 01:41, Kathula, Sandeep wrote:
>
> Hi Jan,
>
> We are not adding any custom timestamp policy. We also don’t see
> backpressure in Flink UI. We are giving 5 task slots each with 3 CPU
> and 32 GB RAM. Its working if we give 10 task slots each with 3 CPU
> and 32 GB RAM. But that’s lot of resources for this load. We are
> trying to figure out why Beam is not able to handle 10,000 records per
> second with 5 task slots.
>
> Thanks,
>
> Sandeep
>
> *From: *Jan Lukavský <je...@seznam.cz>
> *Reply-To: *"user@beam.apache.org" <us...@beam.apache.org>
> *Date: *Tuesday, November 16, 2021 at 3:11 AM
> *To: *"user@beam.apache.org" <us...@beam.apache.org>
> *Subject: *Re: Beam on Flink runner not able to advance watermarks on
> a high load
>
> This email is from an external sender.
>
> Hi Sandeep,
>
> - dev@beam <ma...@beam.apache.org>
>
> The watermark estimation itself should not be related to load. Can you
> please clarify, if
>
> a) you are using any custom timestamp policy?
>
> b) you see any backpressure in Flink's UI? Backpressure could - under
> some circumstances - cause delays in watermark propagation. It _might_
> help to increase parallelism in that case.
>
> Best,
>
> Jan
>
> On 11/15/21 18:22, Kathula, Sandeep wrote:
>
> Hi,
>
> We are running a Beam application on Flink runner (Beam 2.29
> and Flink 1.12) which reads from Kafka and writes to S3 once
> every 5 minutes. My window and s3 writes looks like
>
> PCollection<GenericRecord>.apply("Batch Events",
> Window.<GenericRecord>into(
>
> FixedWindows.of(Duration.standardMinutes(5)))
>
> .triggering(AfterWatermark.pastEndOfWindow())
>
> .withAllowedLateness(Duration.ZERO,
> Window.ClosingBehavior.FIRE_ALWAYS)
>
> .discardingFiredPanes())
>
> .apply(FileIO.<GenericRecord>write()
>
> .via(ParquetIO.sink(schema))
>
> .to(outputPath)
>
> .withNumShards(5)
>
> .withNaming(new CustomFileNaming("snappy.parquet")));
>
> Resources allocated: 5 task slots each with 3 CPU and 32 GB RAM.
> We are using RocksDB as state backend and giving 50% of memory to
> off-heap.
>
> Its running fine with lighter loads. But when it gets heavier load
> from Kafka (7500 or more records per sec – each record around 7KB
> in size), we are seeing that no files are being written to S3.We
> are using AfterWatermark.pastEndOfWindow() which is trigerring
> only when the watermark reaches the end of window.
>
> After debugging we found that watermarks are not being advanced
> during heavy loads and as a result event time triggers after
> watermark reaches end of window because of which s3 writes will
> happen are not getting triggered. So the data is accumulating in
> off-heap which results in out of memory after some time.
>
> Can you please let us know why watermarks are not advancing under
> high load.
>
> Thanks,
>
> Sandeep
>
Re: Beam on Flink runner not able to advance watermarks on a high load
Posted by "Kathula, Sandeep" <Sa...@intuit.com>.
Hi Jan,
We are not adding any custom timestamp policy. We also don’t see backpressure in Flink UI. We are giving 5 task slots each with 3 CPU and 32 GB RAM. Its working if we give 10 task slots each with 3 CPU and 32 GB RAM. But that’s lot of resources for this load. We are trying to figure out why Beam is not able to handle 10,000 records per second with 5 task slots.
Thanks,
Sandeep
From: Jan Lukavský <je...@seznam.cz>
Reply-To: "user@beam.apache.org" <us...@beam.apache.org>
Date: Tuesday, November 16, 2021 at 3:11 AM
To: "user@beam.apache.org" <us...@beam.apache.org>
Subject: Re: Beam on Flink runner not able to advance watermarks on a high load
This email is from an external sender.
Hi Sandeep,
- dev@beam<ma...@beam.apache.org>
The watermark estimation itself should not be related to load. Can you please clarify, if
a) you are using any custom timestamp policy?
b) you see any backpressure in Flink's UI? Backpressure could - under some circumstances - cause delays in watermark propagation. It _might_ help to increase parallelism in that case.
Best,
Jan
On 11/15/21 18:22, Kathula, Sandeep wrote:
Hi,
We are running a Beam application on Flink runner (Beam 2.29 and Flink 1.12) which reads from Kafka and writes to S3 once every 5 minutes. My window and s3 writes looks like
PCollection<GenericRecord>.apply("Batch Events", Window.<GenericRecord>into(
FixedWindows.of(Duration.standardMinutes(5)))
.triggering(AfterWatermark.pastEndOfWindow())
.withAllowedLateness(Duration.ZERO, Window.ClosingBehavior.FIRE_ALWAYS)
.discardingFiredPanes())
.apply(FileIO.<GenericRecord>write()
.via(ParquetIO.sink(schema))
.to(outputPath)
.withNumShards(5)
.withNaming(new CustomFileNaming("snappy.parquet")));
Resources allocated: 5 task slots each with 3 CPU and 32 GB RAM. We are using RocksDB as state backend and giving 50% of memory to off-heap.
Its running fine with lighter loads. But when it gets heavier load from Kafka (7500 or more records per sec – each record around 7KB in size), we are seeing that no files are being written to S3.We are using AfterWatermark.pastEndOfWindow() which is trigerring only when the watermark reaches the end of window.
After debugging we found that watermarks are not being advanced during heavy loads and as a result event time triggers after watermark reaches end of window because of which s3 writes will happen are not getting triggered. So the data is accumulating in off-heap which results in out of memory after some time.
Can you please let us know why watermarks are not advancing under high load.
Thanks,
Sandeep
Re: Beam on Flink runner not able to advance watermarks on a high load
Posted by Jan Lukavský <je...@seznam.cz>.
Hi Sandeep,
- dev@beam <ma...@beam.apache.org>
The watermark estimation itself should not be related to load. Can you
please clarify, if
a) you are using any custom timestamp policy?
b) you see any backpressure in Flink's UI? Backpressure could - under
some circumstances - cause delays in watermark propagation. It _might_
help to increase parallelism in that case.
Best,
Jan
On 11/15/21 18:22, Kathula, Sandeep wrote:
>
> Hi,
>
> We are running a Beam application on Flink runner (Beam 2.29 and
> Flink 1.12) which reads from Kafka and writes to S3 once every 5
> minutes. My window and s3 writes looks like
>
> PCollection<GenericRecord>.apply("Batch Events",
> Window.<GenericRecord>into(
>
> FixedWindows.of(Duration.standardMinutes(5)))
>
> .triggering(AfterWatermark.pastEndOfWindow())
>
> .withAllowedLateness(Duration.ZERO, Window.ClosingBehavior.FIRE_ALWAYS)
>
> .discardingFiredPanes())
>
> .apply(FileIO.<GenericRecord>write()
>
> .via(ParquetIO.sink(schema))
>
> .to(outputPath)
>
> .withNumShards(5)
>
> .withNaming(new CustomFileNaming("snappy.parquet")));
>
> Resources allocated: 5 task slots each with 3 CPU and 32 GB RAM. We
> are using RocksDB as state backend and giving 50% of memory to off-heap.
>
> Its running fine with lighter loads. But when it gets heavier load
> from Kafka (7500 or more records per sec – each record around 7KB in
> size), we are seeing that no files are being written to S3.We are
> using AfterWatermark.pastEndOfWindow() which is trigerring only when
> the watermark reaches the end of window.
>
> After debugging we found that watermarks are not being advanced during
> heavy loads and as a result event time triggers after watermark
> reaches end of window because of which s3 writes will happen are not
> getting triggered. So the data is accumulating in off-heap which
> results in out of memory after some time.
>
> Can you please let us know why watermarks are not advancing under high
> load.
>
> Thanks,
>
> Sandeep
>