[BEAM-9855] Provide an option to configure the Flink state backend #13116

mxm · 2020-10-14T19:44:48Z

We should make it easier to configure a Flink state backend. At the moment,
users have to either:

(A) Configure the default state backend in their Flink cluster

(B1) Include the dependency in their Gradle/Maven
project (e.g. "org.apache.flink:flink-statebackend-rocksdb_2.11:$flink_version"
for RocksDB).
(B2) Set the state backend factory in the FlinkPipelineOptions. This only works
in Java due to the factory specification being in Java!

We can make it easier by simple adding pipeline options for the state backend
name and the checkpoint directory which will be enough for configuring the state
backend. We bundle the RocksDB state backend by default.

Post-Commit Tests Status (on master branch)

Lang	Dataflow	Samza	Twister2
Go	---	---	---
Java
Python		---	---
XLang	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

mxm · 2020-10-16T10:03:06Z

Run Java PreCommit

mxm · 2020-10-16T14:26:35Z

Unrelated flaky tests in Java PreCommit.

tweise · 2020-10-17T21:52:54Z

runners/flink/flink_runner.gradle

@@ -148,6 +148,8 @@ dependencies {
  compile "org.apache.flink:flink-java:$flink_version"
  compile "org.apache.flink:flink-runtime_2.11:$flink_version"
  compile "org.apache.flink:flink-streaming-java_2.11:$flink_version"
+  // RocksDB state backend
+  compile "org.apache.flink:flink-statebackend-rocksdb_2.11:$flink_version"


Just wanted to confirm that dependency won't be baked into the job server as it is already part of the Flink dist.

Is that true? AFAIK RocksDB is an optional dependency. We can change the scope to provided and add a separate dependency with testCompile.

I checked and the dependency does not seem to be part of the default Flink distribution (quite surprising for me btw). However the way of instantiation makes the dependency needed, making it provided will solve it but it is a bit less user-friendly.

The dependency is part of the flink distribution:

$ tar -tf flink-dist_2.12-1.11.2.jar | grep RocksDBStateBackend org/apache/flink/contrib/streaming/state/RocksDBStateBackend.class org/apache/flink/contrib/streaming/state/RocksDBStateBackendFactory.class org/apache/flink/contrib/streaming/state/RocksDBStateBackend$PriorityQueueStateType.class

If it wasn't, you could not configure the default state backend:
https://ci.apache.org/projects/flink/flink-docs-stable/ops/state/state_backends.html#setting-default-state-backend

RocksDB used to be not included in the dist. Good point, we can set it to provided then.

I've verified that RocksDB is not included in any of the to-be-released jars.

tweise · 2020-10-17T21:58:05Z

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

-
-    // State backend
-    if (options.getStateBackendFactory() != null) {
+  private static void configureStateBackend(


Flink is headed in the direction where everything that is set on an environment becomes configurable (including the executor, FLIP-73). This change kind of goes in the opposite direction, increasing the amount of Flink pipeline options further. Should we look into the generic configuration mechanism instead, where it is really easy for the user to supply the Flink configuration (optionally inline, instead of via a separate file)?

I think that's a great idea. We can start thinking about that on the mailing list and handle this via separate JIRA issue. This pipeline option can then be replaced by the generic configuration option.

For now, this pipeline option will fulfill a common request by Beam users to directly set the state backend without having to change the Flink configuration.

Plus, we want to be able to easily run Nexmark with RocksDB.

Is the intention of supporting this to be able to configure new Backends too? I mean like the new one by the RISE team? Is this the intended Nexmark use?

Yes, new state backends can be added as needed.

@iemejia why does Nexmark need the rocksdb state backend? That backend is required when state does not fit into the heap, otherwise it is almost always better to use filesystem.

We may want to experiment with different state backends.

iemejia

I suppose this change is required to specify the backend per job only and there seems not to be another way to do it in current Flink runner. It is curious I expected users to be able to do Flink specific runtime configuration independently of Beam's, isn't there a way to do this?

runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java

website/www/site/layouts/shortcodes/flink_java_pipeline_options.html

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

iemejia · 2020-10-21T14:17:22Z

runners/flink/flink_runner.gradle

@@ -148,6 +148,8 @@ dependencies {
  compile "org.apache.flink:flink-java:$flink_version"
  compile "org.apache.flink:flink-runtime_2.11:$flink_version"
  compile "org.apache.flink:flink-streaming-java_2.11:$flink_version"
+  // RocksDB state backend
+  compile "org.apache.flink:flink-statebackend-rocksdb_2.11:$flink_version"


I checked and the dependency does not seem to be part of the default Flink distribution (quite surprising for me btw). However the way of instantiation makes the dependency needed, making it provided will solve it but it is a bit less user-friendly.

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

iemejia · 2020-10-22T08:03:30Z

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

-
-    // State backend
-    if (options.getStateBackendFactory() != null) {
+  private static void configureStateBackend(


Is the intention of supporting this to be able to configure new Backends too? I mean like the new one by the RISE team? Is this the intended Nexmark use?

mxm

I'll address the remaining comments and will merge this if there are no further remarks.

tweise · 2020-10-27T05:06:23Z

I suppose this change is required to specify the backend per job only and there seems not to be another way to do it in current Flink runner. It is curious I expected users to be able to do Flink specific runtime configuration independently of Beam's, isn't there a way to do this?

I'm not sure that is the case, have you tried it? Can the state backend not be configured as part of flinkConfiguration (

beam/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

Line 155 in 5675108

getDefaultLocalParallelism(), flinkConfiguration);

)?

mxm · 2020-11-08T10:43:05Z

There is currently no universal way to set the configuration per-job. Unless a per-job cluster is used. In this case a configuration file can be supplied for the per-job cluster. There is no way to explicitly configure a state backend via a flag to the main program. Even if we allowed for a configuration file parameter, we would rely on that configuration file to be present during runtime which we can't assume. Further, if we allowed a direct configuration YAML string, we would also have to explicitly set the state backend on the StreamExecutionEnvironment, otherwise during runtime the cluster's default state backend would be used instead of the configured one (the default state backend can only be configured during job creation time). The intention here is to allow setting and overriding the default state backend. Thus, it makes sense (for now) to have an explicit way to configure the state backend.

We should make it easier to configure a Flink state backend. At the moment, users have to either: (A) Configure the default state backend in their Flink cluster (B1) Include the dependency in their Gradle/Maven project (e.g. "org.apache.flink:flink-statebackend-rocksdb_2.11:$flink_version" for RocksDB). (B2) Set the state backend factory in the FlinkPipelineOptions. This only works in Java due to the factory specification being in Java! We can make it easier by simple adding pipeline options for the state backend name and the checkpoint directory which will be enough for configuring the state backend. We bundle the RocksDB state backend by default.

TheNeuralBit · 2020-11-10T23:34:01Z

I think this has broken Java PreCommit (https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/3471/), probably just a missing import

10:18:56 > Task :runners:flink:1.10:compileTestJava
10:18:56 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Cron/src/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java:474: error: cannot find symbol
10:18:56     FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
10:18:56                                    ^
10:18:56   symbol:   variable PipelineOptionsFactory
10:18:56   location: class FlinkExecutionEnvironmentsTest
10:18:56 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Cron/src/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java:489: error: cannot find symbol
10:18:56     FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
10:18:56                                    ^
10:18:56   symbol:   variable PipelineOptionsFactory
10:18:56   location: class FlinkExecutionEnvironmentsTest
10:18:56 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Cron/src/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java:503: error: cannot find symbol
10:18:56     FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
10:18:56                                    ^
10:18:56   symbol:   variable PipelineOptionsFactory
10:18:56   location: class FlinkExecutionEnvironmentsTest
10:18:56 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Cron/src/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java:517: error: cannot find symbol
10:18:56     FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
10:18:56                                    ^
10:18:56   symbol:   variable PipelineOptionsFactory
10:18:56   location: class FlinkExecutionEnvironmentsTest

and #13240.

mxm · 2020-11-11T10:05:22Z

Thanks Brian. Yes, this caused a merge conflict which was not visible before merge. Should have rebased since this PR was open for a bit. Looks like #13297 fixed it.

Refactor checkpointing configuration code

a7a2e01

mxm requested review from sunjincheng121 and tweise October 14, 2020 19:45

mxm force-pushed the BEAM-9855 branch from ee9c3d4 to 7601377 Compare October 14, 2020 20:00

mxm requested a review from iemejia October 16, 2020 08:06

tweise reviewed Oct 17, 2020

View reviewed changes

iemejia reviewed Oct 22, 2020

View reviewed changes

mxm commented Oct 26, 2020

View reviewed changes

mxm force-pushed the BEAM-9855 branch from 7601377 to f292ca8 Compare November 8, 2020 11:58

mxm force-pushed the BEAM-9855 branch from f292ca8 to 74663e2 Compare November 8, 2020 11:59

mxm merged commit d492371 into apache:master Nov 10, 2020

ibzib added a commit to ibzib/beam that referenced this pull request Nov 11, 2020

[BEAM-9855] Fix merge conflict between apache#13116 and apache#13240.

80952c5

udim added a commit that referenced this pull request Nov 11, 2020

Merge pull request #13297: [BEAM-9855] Fix merge conflict between #13116

2339ffc

and #13240.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-9855] Provide an option to configure the Flink state backend #13116

[BEAM-9855] Provide an option to configure the Flink state backend #13116

mxm commented Oct 14, 2020

mxm commented Oct 16, 2020

mxm commented Oct 16, 2020

tweise Oct 17, 2020

mxm Oct 19, 2020

iemejia Oct 21, 2020

tweise Oct 27, 2020

mxm Nov 8, 2020

mxm Nov 10, 2020

tweise Oct 17, 2020

mxm Oct 19, 2020

mxm Oct 19, 2020

iemejia Oct 22, 2020

mxm Oct 26, 2020

tweise Oct 27, 2020

mxm Nov 8, 2020

iemejia left a comment

iemejia Oct 21, 2020

iemejia Oct 22, 2020

mxm left a comment

tweise commented Oct 27, 2020 •

edited

mxm commented Nov 8, 2020

TheNeuralBit commented Nov 10, 2020

mxm commented Nov 11, 2020

[BEAM-9855] Provide an option to configure the Flink state backend #13116

[BEAM-9855] Provide an option to configure the Flink state backend #13116

Conversation

mxm commented Oct 14, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

mxm commented Oct 16, 2020

mxm commented Oct 16, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iemejia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mxm left a comment

Choose a reason for hiding this comment

tweise commented Oct 27, 2020 • edited

mxm commented Nov 8, 2020

TheNeuralBit commented Nov 10, 2020

mxm commented Nov 11, 2020

tweise commented Oct 27, 2020 •

edited