You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2023/07/22 12:13:43 UTC
[beam] branch master updated: [Feature Request]: Support detached mode for flink runner (#26158) (#26193)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 0510dd7c1f0 [Feature Request]: Support detached mode for flink runner (#26158) (#26193)
0510dd7c1f0 is described below
commit 0510dd7c1f0325c19bd241234b2d635716a31eb3
Author: Jeff Zhang <zj...@apache.org>
AuthorDate: Sat Jul 22 20:13:35 2023 +0800
[Feature Request]: Support detached mode for flink runner (#26158) (#26193)
---
.../runners/flink/FlinkDetachedRunnerResult.java | 90 ++++++++++++++++++++--
.../runners/flink/FlinkExecutionEnvironments.java | 25 ++++++
.../flink/FlinkPipelineExecutionEnvironment.java | 49 +++++++++++-
.../beam/runners/flink/FlinkPipelineOptions.java | 14 ++++
.../runners/flink/FlinkPortableRunnerResult.java | 35 ++++++++-
.../org/apache/beam/runners/flink/FlinkRunner.java | 35 +--------
6 files changed, 204 insertions(+), 44 deletions(-)
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkDetachedRunnerResult.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkDetachedRunnerResult.java
index 2f289968cf6..ddf8a36a4f4 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkDetachedRunnerResult.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkDetachedRunnerResult.java
@@ -18,9 +18,16 @@
package org.apache.beam.runners.flink;
import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.runners.flink.metrics.FlinkMetricContainer;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.core.execution.JobClient;
import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Result of a detached execution of a {@link org.apache.beam.sdk.Pipeline} with Flink. In detached
@@ -28,31 +35,102 @@ import org.joda.time.Duration;
*/
public class FlinkDetachedRunnerResult implements PipelineResult {
- FlinkDetachedRunnerResult() {}
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkDetachedRunnerResult.class);
+
+ private JobClient jobClient;
+ private int jobCheckIntervalInSecs;
+
+ FlinkDetachedRunnerResult(JobClient jobClient, int jobCheckIntervalInSecs) {
+ this.jobClient = jobClient;
+ this.jobCheckIntervalInSecs = jobCheckIntervalInSecs;
+ }
@Override
public State getState() {
- return State.UNKNOWN;
+ try {
+ return toBeamJobState(jobClient.getJobStatus().get());
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Fail to get flink job state", e);
+ }
}
@Override
public MetricResults metrics() {
- throw new UnsupportedOperationException("The FlinkRunner does not currently support metrics.");
+ return MetricsContainerStepMap.asAttemptedOnlyMetricResults(getMetricsContainerStepMap());
+ }
+
+ private MetricsContainerStepMap getMetricsContainerStepMap() {
+ try {
+ return (MetricsContainerStepMap)
+ jobClient
+ .getAccumulators()
+ .get()
+ .getOrDefault(FlinkMetricContainer.ACCUMULATOR_NAME, new MetricsContainerStepMap());
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn("Fail to get flink job accumulators", e);
+ return new MetricsContainerStepMap();
+ }
}
@Override
public State cancel() throws IOException {
- throw new UnsupportedOperationException("Cancelling is not yet supported.");
+ try {
+ this.jobClient.cancel().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Fail to cancel flink job", e);
+ }
+ return getState();
}
@Override
public State waitUntilFinish() {
- return State.UNKNOWN;
+ return waitUntilFinish(Duration.millis(Long.MAX_VALUE));
}
@Override
public State waitUntilFinish(Duration duration) {
- return State.UNKNOWN;
+ long start = System.currentTimeMillis();
+ long durationInMillis = duration.getMillis();
+ State state = State.UNKNOWN;
+ while ((System.currentTimeMillis() - start) < durationInMillis) {
+ state = getState();
+ if (state.isTerminal()) {
+ return state;
+ }
+ try {
+ Thread.sleep(jobCheckIntervalInSecs * 1000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (state != null && !state.isTerminal()) {
+ LOG.warn("Job is not finished in {} seconds", duration.getStandardSeconds());
+ }
+ return state;
+ }
+
+ private State toBeamJobState(JobStatus flinkJobStatus) {
+ switch (flinkJobStatus) {
+ case CANCELLING:
+ case CREATED:
+ case INITIALIZING:
+ case FAILING:
+ case RECONCILING:
+ case RESTARTING:
+ case RUNNING:
+ return State.RUNNING;
+ case FINISHED:
+ return State.DONE;
+ case CANCELED:
+ return State.CANCELLED;
+ case FAILED:
+ return State.FAILED;
+ case SUSPENDED:
+ return State.STOPPED;
+ default:
+ return State.UNKNOWN;
+ }
}
@Override
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java
index fb4d90884d2..f0584bf531b 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java
@@ -40,8 +40,10 @@ import org.apache.flink.api.common.ExecutionMode;
import org.apache.flink.api.java.CollectionEnvironment;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.LocalEnvironment;
+import org.apache.flink.api.java.RemoteEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
@@ -128,6 +130,17 @@ public class FlinkExecutionEnvironments {
if (options.getParallelism() != -1 && !(flinkBatchEnv instanceof CollectionEnvironment)) {
flinkBatchEnv.setParallelism(options.getParallelism());
}
+
+ // Only RemoteEnvironment support detached mode, other batch environment enforce to use attached
+ // mode
+ if (!options.getAttachedMode()) {
+ if (flinkBatchEnv instanceof RemoteEnvironment) {
+ flinkBatchEnv.getConfiguration().set(DeploymentOptions.ATTACHED, options.getAttachedMode());
+ } else {
+ LOG.warn("Detached mode is only supported in RemoteEnvironment for batch");
+ }
+ }
+
// Set the correct parallelism, required by UnboundedSourceWrapper to generate consistent
// splits.
final int parallelism;
@@ -238,6 +251,18 @@ public class FlinkExecutionEnvironments {
flinkStreamEnv.disableOperatorChaining();
}
+ // Only RemoteStreamEnvironment support detached mode, other stream environment enforce to use
+ // attached mode.
+ if (!options.getAttachedMode()) {
+ if (flinkStreamEnv instanceof RemoteStreamEnvironment) {
+ ((RemoteStreamEnvironment) flinkStreamEnv)
+ .getClientConfiguration()
+ .set(DeploymentOptions.ATTACHED, options.getAttachedMode());
+ } else {
+ LOG.warn("Detached mode is only supported in RemoteStreamEnvironment for streaming");
+ }
+ }
+
// default to event time
flinkStreamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironment.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironment.java
index 07bb3e9d661..b0ea3764633 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironment.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironment.java
@@ -19,11 +19,16 @@ package org.apache.beam.runners.flink;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+import java.util.Map;
import org.apache.beam.runners.core.construction.resources.PipelineResources;
+import org.apache.beam.runners.core.metrics.MetricsPusher;
import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.metrics.MetricsOptions;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamGraph;
@@ -127,18 +132,56 @@ class FlinkPipelineExecutionEnvironment {
}
/** Launches the program execution. */
- public JobExecutionResult executePipeline() throws Exception {
+ public PipelineResult executePipeline() throws Exception {
final String jobName = options.getJobName();
if (flinkBatchEnv != null) {
- return flinkBatchEnv.execute(jobName);
+ if (options.getAttachedMode()) {
+ JobExecutionResult jobExecutionResult = flinkBatchEnv.execute(jobName);
+ return createAttachedPipelineResult(jobExecutionResult);
+ } else {
+ JobClient jobClient = flinkBatchEnv.executeAsync(jobName);
+ return createDetachedPipelineResult(jobClient, options);
+ }
} else if (flinkStreamEnv != null) {
- return flinkStreamEnv.execute(jobName);
+ if (options.getAttachedMode()) {
+ JobExecutionResult jobExecutionResult = flinkStreamEnv.execute(jobName);
+ return createAttachedPipelineResult(jobExecutionResult);
+ } else {
+ JobClient jobClient = flinkStreamEnv.executeAsync(jobName);
+ return createDetachedPipelineResult(jobClient, options);
+ }
} else {
throw new IllegalStateException("The Pipeline has not yet been translated.");
}
}
+ private FlinkDetachedRunnerResult createDetachedPipelineResult(
+ JobClient jobClient, FlinkPipelineOptions options) {
+ LOG.info("Pipeline submitted in detached mode");
+ return new FlinkDetachedRunnerResult(jobClient, options.getJobCheckIntervalInSecs());
+ }
+
+ private FlinkRunnerResult createAttachedPipelineResult(JobExecutionResult result) {
+ LOG.info("Execution finished in {} msecs", result.getNetRuntime());
+ Map<String, Object> accumulators = result.getAllAccumulatorResults();
+ if (accumulators != null && !accumulators.isEmpty()) {
+ LOG.info("Final accumulator values:");
+ for (Map.Entry<String, Object> entry : result.getAllAccumulatorResults().entrySet()) {
+ LOG.info("{} : {}", entry.getKey(), entry.getValue());
+ }
+ }
+ FlinkRunnerResult flinkRunnerResult =
+ new FlinkRunnerResult(accumulators, result.getNetRuntime());
+ MetricsPusher metricsPusher =
+ new MetricsPusher(
+ flinkRunnerResult.getMetricsContainerStepMap(),
+ options.as(MetricsOptions.class),
+ flinkRunnerResult);
+ metricsPusher.start();
+ return flinkRunnerResult;
+ }
+
/**
* Retrieves the generated JobGraph which can be submitted against the cluster. For testing
* purposes.
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java
index 02d4a895cdd..2f32ab2c2ea 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java
@@ -142,6 +142,20 @@ public interface FlinkPipelineOptions
void setNumberOfExecutionRetries(Integer retries);
+ @Description(
+ "Set job check interval in seconds under detached mode in method waitUntilFinish, "
+ + "by default it is 5 seconds")
+ @Default.Integer(5)
+ int getJobCheckIntervalInSecs();
+
+ void setJobCheckIntervalInSecs(int seconds);
+
+ @Description("Specifies if the pipeline is submitted in attached or detached mode")
+ @Default.Boolean(true)
+ boolean getAttachedMode();
+
+ void setAttachedMode(boolean attachedMode);
+
@Description(
"Sets the delay in milliseconds between executions. A value of {@code -1} "
+ "indicates that the default value should be used.")
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableRunnerResult.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableRunnerResult.java
index f9b697708b1..de19f3dfe74 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableRunnerResult.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableRunnerResult.java
@@ -17,10 +17,13 @@
*/
package org.apache.beam.runners.flink;
+import java.io.IOException;
import java.util.Map;
import org.apache.beam.model.jobmanagement.v1.JobApi;
import org.apache.beam.model.pipeline.v1.MetricsApi;
import org.apache.beam.runners.jobsubmission.PortablePipelineResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,7 +47,11 @@ public class FlinkPortableRunnerResult extends FlinkRunnerResult implements Port
.build();
}
- static class Detached extends FlinkDetachedRunnerResult implements PortablePipelineResult {
+ static class Detached implements PortablePipelineResult {
+
+ Detached() {
+ super();
+ }
@Override
public JobApi.MetricResults portableMetrics() throws UnsupportedOperationException {
@@ -52,5 +59,31 @@ public class FlinkPortableRunnerResult extends FlinkRunnerResult implements Port
"Collecting monitoring infos is not implemented yet in Flink portable runner (detached mode).");
return JobApi.MetricResults.newBuilder().build();
}
+
+ @Override
+ public State getState() {
+ return State.UNKNOWN;
+ }
+
+ @Override
+ public State cancel() throws IOException {
+ throw new UnsupportedOperationException("Cancelling is not yet supported.");
+ }
+
+ @Override
+ public State waitUntilFinish(Duration duration) {
+ return State.UNKNOWN;
+ }
+
+ @Override
+ public State waitUntilFinish() {
+ return State.UNKNOWN;
+ }
+
+ @Override
+ public MetricResults metrics() {
+ throw new UnsupportedOperationException(
+ "The FlinkRunner does not currently support metrics.");
+ }
}
}
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
index d047e17eba5..b2403f6f5b5 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
@@ -18,18 +18,15 @@
package org.apache.beam.runners.flink;
import java.util.HashSet;
-import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import org.apache.beam.runners.core.construction.SplittableParDo;
import org.apache.beam.runners.core.construction.graph.ProjectionPushdownOptimizer;
-import org.apache.beam.runners.core.metrics.MetricsPusher;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.PipelineRunner;
import org.apache.beam.sdk.metrics.MetricsEnvironment;
-import org.apache.beam.sdk.metrics.MetricsOptions;
import org.apache.beam.sdk.options.ExperimentalOptions;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsValidator;
@@ -37,7 +34,6 @@ import org.apache.beam.sdk.runners.TransformHierarchy;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
-import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -103,42 +99,13 @@ public class FlinkRunner extends PipelineRunner<PipelineResult> {
LOG.info("Translating pipeline to Flink program.");
env.translate(pipeline);
- JobExecutionResult result;
try {
LOG.info("Starting execution of Flink program.");
- result = env.executePipeline();
+ return env.executePipeline();
} catch (Exception e) {
LOG.error("Pipeline execution failed", e);
throw new RuntimeException("Pipeline execution failed", e);
}
- return createPipelineResult(result, options);
- }
-
- static PipelineResult createPipelineResult(JobExecutionResult result, PipelineOptions options) {
- String resultClassName = result.getClass().getCanonicalName();
- if (resultClassName.equals("org.apache.flink.core.execution.DetachedJobExecutionResult")) {
- LOG.info("Pipeline submitted in Detached mode");
- // no metricsPusher because metrics are not supported in detached mode
- return new FlinkDetachedRunnerResult();
- } else {
- LOG.info("Execution finished in {} msecs", result.getNetRuntime());
- Map<String, Object> accumulators = result.getAllAccumulatorResults();
- if (accumulators != null && !accumulators.isEmpty()) {
- LOG.info("Final accumulator values:");
- for (Map.Entry<String, Object> entry : result.getAllAccumulatorResults().entrySet()) {
- LOG.info("{} : {}", entry.getKey(), entry.getValue());
- }
- }
- FlinkRunnerResult flinkRunnerResult =
- new FlinkRunnerResult(accumulators, result.getNetRuntime());
- MetricsPusher metricsPusher =
- new MetricsPusher(
- flinkRunnerResult.getMetricsContainerStepMap(),
- options.as(MetricsOptions.class),
- flinkRunnerResult);
- metricsPusher.start();
- return flinkRunnerResult;
- }
}
/** For testing. */