You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/10/29 10:39:58 UTC
[flink] branch master updated (2bb4012 -> e6ef948)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 2bb4012 [FLINK-24462][table] Introduce CastRule interface to reorganize casting code
new 421d71e [FLINK-24670][docs] Restructure unaligned checkpoints docs to checkpointing under backpressure
new e6ef948 [FLINK-24695][docs] Update how to configure unaligned checkpoints
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../ops/state/checkpointing_under_backpressure.md} | 85 ++++++++++++++++++++--
.../datastream/fault-tolerance/checkpointing.md | 8 +-
...ints.md => checkpointing_under_backpressure.md} | 85 ++++++++++++++++++++--
3 files changed, 158 insertions(+), 20 deletions(-)
copy docs/{content/docs/ops/state/unaligned_checkpoints.md => content.zh/docs/ops/state/checkpointing_under_backpressure.md} (67%)
rename docs/content/docs/ops/state/{unaligned_checkpoints.md => checkpointing_under_backpressure.md} (67%)
[flink] 01/02: [FLINK-24670][docs] Restructure unaligned
checkpoints docs to checkpointing under backpressure
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 421d71e3d14d2c0b48a7249c72feb7378cc1786a
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Oct 27 17:32:03 2021 +0200
[FLINK-24670][docs] Restructure unaligned checkpoints docs to checkpointing under backpressure
---
.../ops/state/checkpointing_under_backpressure.md} | 50 ++++++++++++++++++----
.../datastream/fault-tolerance/checkpointing.md | 2 +-
...ints.md => checkpointing_under_backpressure.md} | 50 ++++++++++++++++++----
3 files changed, 85 insertions(+), 17 deletions(-)
diff --git a/docs/content/docs/ops/state/unaligned_checkpoints.md b/docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md
similarity index 73%
copy from docs/content/docs/ops/state/unaligned_checkpoints.md
copy to docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md
index 591c838..9226012 100644
--- a/docs/content/docs/ops/state/unaligned_checkpoints.md
+++ b/docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md
@@ -1,9 +1,10 @@
---
-title: "Unaligned checkpoints"
+title: "Checkpointing under backpressure"
weight: 9
type: docs
aliases:
- /ops/state/unalgined_checkpoints.html
+- /ops/state/checkpointing_under_backpressure.html
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -23,7 +24,40 @@ KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
-# Unaligned checkpoints
+# Checkpointing under backpressure
+
+Normally aligned checkpointing time is dominated by the synchronous and asynchronous parts of the
+checkpointing process. However, when a Flink job is running under heavy backpressure, the dominant
+factor in the end-to-end time of a checkpoint can be the time to propagate checkpoint barriers to
+all operators/subtasks. This is explained in the overview of the
+[checkpointing process]({{< ref "docs/concepts/stateful-stream-processing" >}}#checkpointing)).
+and can be observed by high
+[alignment time and start delay metrics]({{< ref "docs/ops/monitoring/checkpoint_monitoring" >}}#history-tab).
+When this happens and becomes an issue, there are three ways to address the problem:
+1. Remove the backpressure source by optimizing the Flink job, by adjusting Flink or JVM configurations, or by scaling up.
+2. Reduce the amount of buffered in-flight data in the Flink job.
+3. Enable unaligned checkpoints.
+
+These options are not mutually exclusive and can be combined together. This document
+focuses on the latter two options.
+
+## Buffer debloating
+
+Flink 1.14 introduced a new tool to automatically control the amount of buffered in-flight data
+between Flink operators/subtasks. The buffer debloating mechanism can be enabled by setting the property
+`taskmanager.network.memory.buffer-debloat.enabled` to `true`.
+
+This feature works with both aligned and unaligned checkpoints and can improve checkpointing times
+in both cases, but the effect of the debloating is most visible with aligned checkpoints.
+When using buffer debloating with unaligned checkpoints, the added benefit will be smaller checkpoint
+sizes and quicker recovery times (there will be less in-flight data to persist and recover).
+
+For more information on how the buffer debloating feature works and how to configure it, please refer to the
+[network memory tuning guide]({{< ref "docs/deployment/memory/network_mem_tuning" >}}).
+Keep in mind that you can also manually reduce the amount of buffered in-flight data which is also
+described in the aforementioned tuning guide.
+
+## Unaligned checkpoints
Starting with Flink 1.11, checkpoints can be unaligned.
[Unaligned checkpoints]({{< ref "docs/concepts/stateful-stream-processing" >}}#unaligned-checkpointing)
@@ -57,16 +91,16 @@ When activated, each checkpoint will still begin as an aligned checkpoint, but i
the start of the global checkpoint and the start of the checkpoint on a subtask exceeds the aligned
checkpoint timeout, then the checkpoint will proceed as an unaligned checkpoint.
-## Limitations
+### Limitations
-### Concurrent checkpoints
+#### Concurrent checkpoints
Flink currently does not support concurrent unaligned checkpoints. However, due to the more
predictable and shorter checkpointing times, concurrent checkpoints might not be needed at all.
However, savepoints can also not happen concurrently to unaligned checkpoints, so they will take
slightly longer.
-### Interplay with watermarks
+#### Interplay with watermarks
Unaligned checkpoints break with an implicit guarantee in respect to watermarks during recovery.
Currently, Flink generates the watermark as the first step of recovery instead of storing the latest
@@ -77,7 +111,7 @@ aligned checkpoints. If your operator depends on the latest watermark being alwa
workaround is to store the watermark in the operator state. In that case, watermarks should be
stored per key group in a union state to support rescaling.
-### Certain data distribution patterns are not checkpointed
+#### Certain data distribution patterns are not checkpointed
There are types of connections with properties that are impossible to keep with channel data stored
in checkpoints. To preserve these characteristics and ensure no state corruption or unexpected
@@ -119,9 +153,9 @@ from subtask 0 of the stateful operator. Upon restore, we send that copy to all
operators. Therefore it might happen that an operator will get the state with changes applied for a
record that it will soon consume from its checkpointed channels.
-## Troubleshooting
+### Troubleshooting
-### Corrupted in-flight data
+#### Corrupted in-flight data
{{< hint warning >}}
Actions described below are a last resort as they will lead to data loss.
{{< /hint >}}
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
index 2708561..936bdb1 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
@@ -81,7 +81,7 @@ Other parameters for checkpointing include:
- *externalized checkpoints*: You can configure periodic checkpoints to be persisted externally. Externalized checkpoints write their meta data out to persistent storage and are *not* automatically cleaned up when the job fails. This way, you will have a checkpoint around to resume from if your job fails. There are more details in the [deployment notes on externalized checkpoints]({{< ref "docs/ops/state/checkpoints" >}}#externalized-checkpoints).
- - *unaligned checkpoints*: You can enable [unaligned checkpoints]({{< ref "docs/ops/state/unaligned_checkpoints" >}}) to greatly reduce checkpointing times under backpressure. This only works for exactly-once checkpoints and with number of concurrent checkpoints of 1.
+ - *unaligned checkpoints*: You can enable [unaligned checkpoints]({{< ref "docs/ops/state/checkpointing_under_backpressure" >}}#unaligned-checkpoints) to greatly reduce checkpointing times under backpressure. This only works for exactly-once checkpoints and with one concurrent checkpoint.
- *checkpoints with finished tasks*: You can enable an experimental feature to continue performing checkpoints even if parts of the DAG have finished processing all of their records. Before doing so, please read through some [important considerations](#checkpointing-with-parts-of-the-graph-finished).
diff --git a/docs/content/docs/ops/state/unaligned_checkpoints.md b/docs/content/docs/ops/state/checkpointing_under_backpressure.md
similarity index 73%
rename from docs/content/docs/ops/state/unaligned_checkpoints.md
rename to docs/content/docs/ops/state/checkpointing_under_backpressure.md
index 591c838..9226012 100644
--- a/docs/content/docs/ops/state/unaligned_checkpoints.md
+++ b/docs/content/docs/ops/state/checkpointing_under_backpressure.md
@@ -1,9 +1,10 @@
---
-title: "Unaligned checkpoints"
+title: "Checkpointing under backpressure"
weight: 9
type: docs
aliases:
- /ops/state/unalgined_checkpoints.html
+- /ops/state/checkpointing_under_backpressure.html
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -23,7 +24,40 @@ KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
-# Unaligned checkpoints
+# Checkpointing under backpressure
+
+Normally aligned checkpointing time is dominated by the synchronous and asynchronous parts of the
+checkpointing process. However, when a Flink job is running under heavy backpressure, the dominant
+factor in the end-to-end time of a checkpoint can be the time to propagate checkpoint barriers to
+all operators/subtasks. This is explained in the overview of the
+[checkpointing process]({{< ref "docs/concepts/stateful-stream-processing" >}}#checkpointing)).
+and can be observed by high
+[alignment time and start delay metrics]({{< ref "docs/ops/monitoring/checkpoint_monitoring" >}}#history-tab).
+When this happens and becomes an issue, there are three ways to address the problem:
+1. Remove the backpressure source by optimizing the Flink job, by adjusting Flink or JVM configurations, or by scaling up.
+2. Reduce the amount of buffered in-flight data in the Flink job.
+3. Enable unaligned checkpoints.
+
+These options are not mutually exclusive and can be combined together. This document
+focuses on the latter two options.
+
+## Buffer debloating
+
+Flink 1.14 introduced a new tool to automatically control the amount of buffered in-flight data
+between Flink operators/subtasks. The buffer debloating mechanism can be enabled by setting the property
+`taskmanager.network.memory.buffer-debloat.enabled` to `true`.
+
+This feature works with both aligned and unaligned checkpoints and can improve checkpointing times
+in both cases, but the effect of the debloating is most visible with aligned checkpoints.
+When using buffer debloating with unaligned checkpoints, the added benefit will be smaller checkpoint
+sizes and quicker recovery times (there will be less in-flight data to persist and recover).
+
+For more information on how the buffer debloating feature works and how to configure it, please refer to the
+[network memory tuning guide]({{< ref "docs/deployment/memory/network_mem_tuning" >}}).
+Keep in mind that you can also manually reduce the amount of buffered in-flight data which is also
+described in the aforementioned tuning guide.
+
+## Unaligned checkpoints
Starting with Flink 1.11, checkpoints can be unaligned.
[Unaligned checkpoints]({{< ref "docs/concepts/stateful-stream-processing" >}}#unaligned-checkpointing)
@@ -57,16 +91,16 @@ When activated, each checkpoint will still begin as an aligned checkpoint, but i
the start of the global checkpoint and the start of the checkpoint on a subtask exceeds the aligned
checkpoint timeout, then the checkpoint will proceed as an unaligned checkpoint.
-## Limitations
+### Limitations
-### Concurrent checkpoints
+#### Concurrent checkpoints
Flink currently does not support concurrent unaligned checkpoints. However, due to the more
predictable and shorter checkpointing times, concurrent checkpoints might not be needed at all.
However, savepoints can also not happen concurrently to unaligned checkpoints, so they will take
slightly longer.
-### Interplay with watermarks
+#### Interplay with watermarks
Unaligned checkpoints break with an implicit guarantee in respect to watermarks during recovery.
Currently, Flink generates the watermark as the first step of recovery instead of storing the latest
@@ -77,7 +111,7 @@ aligned checkpoints. If your operator depends on the latest watermark being alwa
workaround is to store the watermark in the operator state. In that case, watermarks should be
stored per key group in a union state to support rescaling.
-### Certain data distribution patterns are not checkpointed
+#### Certain data distribution patterns are not checkpointed
There are types of connections with properties that are impossible to keep with channel data stored
in checkpoints. To preserve these characteristics and ensure no state corruption or unexpected
@@ -119,9 +153,9 @@ from subtask 0 of the stateful operator. Upon restore, we send that copy to all
operators. Therefore it might happen that an operator will get the state with changes applied for a
record that it will soon consume from its checkpointed channels.
-## Troubleshooting
+### Troubleshooting
-### Corrupted in-flight data
+#### Corrupted in-flight data
{{< hint warning >}}
Actions described below are a last resort as they will lead to data loss.
{{< /hint >}}
[flink] 02/02: [FLINK-24695][docs] Update how to configure
unaligned checkpoints
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e6ef94814e4730bf9762274e52e37c5c4d0d0b9e
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Oct 29 09:39:53 2021 +0200
[FLINK-24695][docs] Update how to configure unaligned checkpoints
---
.../ops/state/checkpointing_under_backpressure.md | 35 ++++++++++++++++++++++
.../datastream/fault-tolerance/checkpointing.md | 6 ++--
.../ops/state/checkpointing_under_backpressure.md | 35 ++++++++++++++++++++++
3 files changed, 73 insertions(+), 3 deletions(-)
diff --git a/docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md b/docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md
index 9226012..14277f4 100644
--- a/docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md
+++ b/docs/content.zh/docs/ops/state/checkpointing_under_backpressure.md
@@ -71,6 +71,41 @@ backpressure. Then, checkpointing time becomes mostly independent of the end-to-
aware unaligned checkpointing adds to I/O to the state storage, so you shouldn't use it when the
I/O to the state storage is actually the bottleneck during checkpointing.
+In order to enable unaligned checkpoints you can:
+
+{{< tabs "4b9c6a74-8a45-4ad2-9e80-52fe44a85991" >}}
+{{< tab "Java" >}}
+```java
+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+// enables the unaligned checkpoints
+env.getCheckpointConfig().enableUnalignedCheckpoints();
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+
+// enables the unaligned checkpoints
+env.getCheckpointConfig.enableUnalignedCheckpoints()
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# enables the unaligned checkpoints
+env.get_checkpoint_config().enable_unaligned_checkpoints()
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+or in the `flink-conf.yml` configuration file:
+
+```
+execution.checkpointing.unaligned: true
+```
+
### Aligned checkpoint timeout
After enabling unaligned checkpoints, you can also specify the aligned checkpoint timeout
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
index 936bdb1..8704582 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/checkpointing.md
@@ -115,7 +115,7 @@ env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().enableExternalizedCheckpoints(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
-// enables the experimental unaligned checkpoints
+// enables the unaligned checkpoints
env.getCheckpointConfig().enableUnalignedCheckpoints();
// sets the checkpoint storage where checkpoint snapshots will be written
@@ -156,7 +156,7 @@ env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
env.getCheckpointConfig().enableExternalizedCheckpoints(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
-// enables the experimental unaligned checkpoints
+// enables the unaligned checkpoints
env.getCheckpointConfig.enableUnalignedCheckpoints()
// sets the checkpoint storage where checkpoint snapshots will be written
@@ -195,7 +195,7 @@ env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
# enable externalized checkpoints which are retained after job cancellation
env.get_checkpoint_config().enable_externalized_checkpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
-# enables the experimental unaligned checkpoints
+# enables the unaligned checkpoints
env.get_checkpoint_config().enable_unaligned_checkpoints()
```
{{< /tab >}}
diff --git a/docs/content/docs/ops/state/checkpointing_under_backpressure.md b/docs/content/docs/ops/state/checkpointing_under_backpressure.md
index 9226012..14277f4 100644
--- a/docs/content/docs/ops/state/checkpointing_under_backpressure.md
+++ b/docs/content/docs/ops/state/checkpointing_under_backpressure.md
@@ -71,6 +71,41 @@ backpressure. Then, checkpointing time becomes mostly independent of the end-to-
aware unaligned checkpointing adds to I/O to the state storage, so you shouldn't use it when the
I/O to the state storage is actually the bottleneck during checkpointing.
+In order to enable unaligned checkpoints you can:
+
+{{< tabs "4b9c6a74-8a45-4ad2-9e80-52fe44a85991" >}}
+{{< tab "Java" >}}
+```java
+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+// enables the unaligned checkpoints
+env.getCheckpointConfig().enableUnalignedCheckpoints();
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+
+// enables the unaligned checkpoints
+env.getCheckpointConfig.enableUnalignedCheckpoints()
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# enables the unaligned checkpoints
+env.get_checkpoint_config().enable_unaligned_checkpoints()
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+or in the `flink-conf.yml` configuration file:
+
+```
+execution.checkpointing.unaligned: true
+```
+
### Aligned checkpoint timeout
After enabling unaligned checkpoints, you can also specify the aligned checkpoint timeout