You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/10/16 12:43:15 UTC
[flink] 04/15: [FLINK-14290] Add support for StreamGraph to
pipeline translation util
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 55eb4b254eb531287ec84c63874b479d42f8e315
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Thu Sep 12 13:37:18 2019 +0200
[FLINK-14290] Add support for StreamGraph to pipeline translation util
---
.../flink/client/FlinkPipelineTranslationUtil.java | 38 ++++++++++++++++++-
.../flink/streaming/api/graph/StreamGraph.java | 3 +-
.../streaming/api/graph/StreamGraphTranslator.java | 43 +++++++++++++---------
3 files changed, 64 insertions(+), 20 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java b/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
index 61bda14..33c8027 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
@@ -47,6 +47,42 @@ public final class FlinkPipelineTranslationUtil {
defaultParallelism);
}
- throw new RuntimeException("Cannot find transmogrifier for given pipeline: " + pipeline);
+ FlinkPipelineTranslator streamGraphTranslator = reflectStreamGraphTranslator();
+
+ if (!streamGraphTranslator.canTranslate(pipeline)) {
+ throw new RuntimeException("Translator " + streamGraphTranslator + " cannot translate "
+ + "the given pipeline " + pipeline + ".");
+ }
+
+ return streamGraphTranslator.translate(pipeline,
+ optimizerConfiguration,
+ defaultParallelism);
+ }
+
+ private static FlinkPipelineTranslator reflectStreamGraphTranslator() {
+ // Try our luck with StreamGraph translation. We have to load a StreamGraphTranslator
+ // via reflection because the dependencies of flink-streaming-java are inverted compared
+ // to flink-java. For flink-java does not depend on runtime, clients or optimizer and
+ // we have the translation code in clients/optimizer. On the other hand,
+ // flink-streaming-java depends on runtime and clients.
+
+ Class<?> streamGraphTranslatorClass;
+ try {
+ streamGraphTranslatorClass = Class.forName(
+ "org.apache.flink.streaming.api.graph.StreamGraphTranslator",
+ true,
+ FlinkPipelineTranslationUtil.class.getClassLoader());
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Could not load StreamGraphTranslator.", e);
+ }
+
+ FlinkPipelineTranslator streamGraphTranslator;
+ try {
+ streamGraphTranslator =
+ (FlinkPipelineTranslator) streamGraphTranslatorClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException e) {
+ throw new RuntimeException("Could not instantiate StreamGraphTranslator.", e);
+ }
+ return streamGraphTranslator;
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index f4741fb..522c6fc 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -71,7 +72,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*
*/
@Internal
-public class StreamGraph extends StreamingPlan {
+public class StreamGraph extends StreamingPlan implements Pipeline {
private static final Logger LOG = LoggerFactory.getLogger(StreamGraph.class);
diff --git a/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphTranslator.java
similarity index 52%
copy from flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
copy to flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphTranslator.java
index 61bda14..ebd554f 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphTranslator.java
@@ -17,36 +17,43 @@
*
*/
-package org.apache.flink.client;
+package org.apache.flink.streaming.api.graph;
import org.apache.flink.api.dag.Pipeline;
+import org.apache.flink.client.FlinkPipelineTranslator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
- * Utility for transforming {@link Pipeline FlinkPipelines} into a {@link JobGraph}. This uses
- * reflection or service discovery to find the right {@link FlinkPipelineTranslator} for a given
- * subclass of {@link Pipeline}.
+ * {@link FlinkPipelineTranslator} for DataStream API {@link StreamGraph StreamGraphs}.
+ *
+ * <p>Note: this is used through reflection in
+ * {@link org.apache.flink.client.FlinkPipelineTranslationUtil}.
*/
-public final class FlinkPipelineTranslationUtil {
+@SuppressWarnings("unused")
+public class StreamGraphTranslator implements FlinkPipelineTranslator {
- /**
- * Transmogrifies the given {@link Pipeline} to a {@link JobGraph}.
- */
- public static JobGraph getJobGraph(
+ private static final Logger LOG = LoggerFactory.getLogger(StreamGraphTranslator.class);
+
+ @Override
+ public JobGraph translate(
Pipeline pipeline,
Configuration optimizerConfiguration,
int defaultParallelism) {
+ checkArgument(pipeline instanceof StreamGraph,
+ "Given pipeline is not a DataStream StreamGraph.");
- PlanTranslator planToJobGraphTransmogrifier = new PlanTranslator();
-
- if (planToJobGraphTransmogrifier.canTranslate(pipeline)) {
- // we have a DataSet program
- return planToJobGraphTransmogrifier.translate(pipeline,
- optimizerConfiguration,
- defaultParallelism);
- }
+ StreamGraph streamGraph = (StreamGraph) pipeline;
+ return streamGraph.getJobGraph();
+ }
- throw new RuntimeException("Cannot find transmogrifier for given pipeline: " + pipeline);
+ @Override
+ public boolean canTranslate(Pipeline pipeline) {
+ return pipeline instanceof StreamGraph;
}
}