You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by XuPingyong <gi...@git.apache.org> on 2017/07/06 15:26:35 UTC
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
GitHub user XuPingyong opened a pull request:
https://github.com/apache/flink/pull/4273
[FLINK-7065] Separate the flink-streaming-java module from flink-clients
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/XuPingyong/flink FLINK-7065
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4273.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4273
----
commit ac3da536a2870a5c5f92170c455862c9b9e133df
Author: pingyong.xpy <pi...@alibaba-inc.com>
Date: 2017-07-04T12:29:17Z
[FLINK-7065] Separate the flink-streaming-java module from flink-clients
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129599728
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/Executor.java ---
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * A Executor execute a Flink program's job.
+ *
+ * <p>The specific implementation (such as the org.apache.flink.client.LocalExecutor
+ * and org.apache.flink.client.RemoteExecutor) determines where and how to run the dataflow.
+ * The concrete implementations of the executors are loaded dynamically, because they depend on
+ * the full set of all runtime classes.</p>
+ */
+@Internal
+public interface Executor {
--- End diff --
Can we rename it? It clashes with `java.util.concurrent.Executor`. Maybe something like `ProgramExecutor` or so.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129599271
--- Diff: flink-contrib/flink-connector-wikiedits/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <artifactId>flink-clients_${scala.binary.version}</artifactId>
--- End diff --
Why does this module need a dependency on `flink-clients`? Is it being executed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4273: [FLINK-7065] Separate the flink-streaming-java module fro...
Posted by XuPingyong <gi...@git.apache.org>.
Github user XuPingyong commented on the issue:
https://github.com/apache/flink/pull/4273
@tillrohrmann , could you please spend some time on reviewing this PR?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4273: [FLINK-7065] Separate the flink-streaming-java module fro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/4273
Hi @XuPingyong, @tillrohrmann is not available for the next two weeks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129598422
--- Diff: flink-clients/src/main/java/org/apache/flink/client/RemoteExecutor.java ---
@@ -106,8 +113,14 @@ public RemoteExecutor(InetSocketAddress inet, Configuration clientConfiguration,
List<URL> jarFiles, List<URL> globalClasspaths) {
this.clientConfiguration = clientConfiguration;
this.jarFiles = jarFiles;
+ for (URL jarFileUrl : jarFiles) {
+ try {
+ JobWithJars.checkJarFile(jarFileUrl);
+ } catch (IOException e) {
+ throw new RuntimeException("Problem with jar file " + jarFileUrl, e);
--- End diff --
Can we make this a checked exception?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129582723
--- Diff: flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java ---
@@ -37,19 +37,22 @@
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+import org.apache.flink.streaming.api.environment.StreamGraphExecutor;
+import org.apache.flink.streaming.api.graph.StreamGraph;
import java.util.List;
/**
- * A PlanExecutor that runs Flink programs on a local embedded Flink runtime instance.
+ * A LocalExecutor that runs Flink programs or streamGraphs on a local embedded Flink runtime instance.
--- End diff --
Capital letter `StreamGraphs`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129598734
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java ---
@@ -37,8 +37,11 @@
private FlinkPlan optimizerPlan;
+ private StreamPlanEnvironment streamPlanEnvironment;
--- End diff --
can this be final?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4273: [FLINK-7065] Separate the flink-streaming-java module fro...
Posted by XuPingyong <gi...@git.apache.org>.
Github user XuPingyong commented on the issue:
https://github.com/apache/flink/pull/4273
Thanks @tillrohrmann , I have updated this PR according to your advice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129582912
--- Diff: flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java ---
@@ -103,6 +109,16 @@ public int getTaskManagerNumSlots() {
// --------------------------------------------------------------------------------------------
@Override
+ public void setPrintStatusDuringExecution(boolean printStatus) {
+ printUpdatesToSysout= printStatus;
--- End diff --
whitespace after variable name
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129598804
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/PreviewPlanEnvironment.java ---
@@ -38,6 +38,12 @@
Plan plan;
+ StreamPlanEnvironment streamPlanEnvironment;
--- End diff --
final?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by XuPingyong <gi...@git.apache.org>.
Github user XuPingyong commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r131916494
--- Diff: flink-contrib/flink-connector-wikiedits/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <artifactId>flink-clients_${scala.binary.version}</artifactId>
--- End diff --
It is executed in WikipediaEditsSourceTest. In the latest commit, this module still depends on flink-streaming-java, and also flink-clients of test scope.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129600397
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/ExecutorFactory.java ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+
+import java.net.URL;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A ExecutorFactory create the specific implementation of {@link Executor}
+ *
+ */
+@Internal
+public class ExecutorFactory<T extends Executor> {
+
+ private static final String LOCAL_EXECUTOR_CLASS = "org.apache.flink.client.LocalExecutor";
+ private static final String REMOTE_EXECUTOR_CLASS = "org.apache.flink.client.RemoteExecutor";
+
+ private Class<T> clazz;
+
+ public ExecutorFactory(Class<T> clazz) {
+ this.clazz = clazz;
+ }
+
+ // ------------------------------------------------------------------------
+ // Executor Factories
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates an executor that runs locally in a multi-threaded environment.
+ *
+ * @return A local executor.
+ */
+ public T createLocalExecutor(Configuration configuration) {
--- End diff --
Can we make this a static method? Then we would not have to create a factory object.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129601152
--- Diff: flink-streaming-java/pom.xml ---
@@ -52,7 +52,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-clients_${scala.binary.version}</artifactId>
+ <artifactId>flink-optimizer_${scala.binary.version}</artifactId>
--- End diff --
can we remove the `flink-optimizer` dependency? Maybe we have to move `FlinkPlan` and `StreamingPlan` out of `flink-optimizer` for that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r129601816
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamGraphExecutor.java ---
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.environment;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.Executor;
+import org.apache.flink.api.common.ExecutorFactory;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+
+/**
+ * A StreamGraphExecutor executes a StreamGraph.
+ *
+ * <p>The specific implementation (such as the org.apache.flink.client.LocalExecutor
+ * and org.apache.flink.client.RemoteExecutor) is created by {@link ExecutorFactory}</p>
+ *
+ */
+@Internal
+public interface StreamGraphExecutor extends Executor {
+
+ /**
+ * Execute the given program.
+ *
+ * <p>If the executor has not been started before, then this method will start the
+ * executor and stop it after the execution has completed. This implies that one needs
+ * to explicitly start the executor for all programs where multiple dataflow parts
+ * depend on each other. Otherwise, the previous parts will no longer
+ * be available, because the executor immediately shut down after the execution.</p>
+ *
+ * @param streamGraph The streamGraph to execute.
+ * @return The execution result, containing for example the net runtime of the program, and the accumulators.
+ *
+ * @throws Exception Thrown, if job submission caused an exception.
+ */
+ JobExecutionResult executeStreamGraph(StreamGraph streamGraph) throws Exception;
--- End diff --
Could the thrown exception be of type `FlinkException`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4273: [FLINK-7065] Separate the flink-streaming-java mod...
Posted by XuPingyong <gi...@git.apache.org>.
Github user XuPingyong commented on a diff in the pull request:
https://github.com/apache/flink/pull/4273#discussion_r131918248
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamGraphExecutor.java ---
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.environment;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.Executor;
+import org.apache.flink.api.common.ExecutorFactory;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+
+/**
+ * A StreamGraphExecutor executes a StreamGraph.
+ *
+ * <p>The specific implementation (such as the org.apache.flink.client.LocalExecutor
+ * and org.apache.flink.client.RemoteExecutor) is created by {@link ExecutorFactory}</p>
+ *
+ */
+@Internal
+public interface StreamGraphExecutor extends Executor {
+
+ /**
+ * Execute the given program.
+ *
+ * <p>If the executor has not been started before, then this method will start the
+ * executor and stop it after the execution has completed. This implies that one needs
+ * to explicitly start the executor for all programs where multiple dataflow parts
+ * depend on each other. Otherwise, the previous parts will no longer
+ * be available, because the executor immediately shut down after the execution.</p>
+ *
+ * @param streamGraph The streamGraph to execute.
+ * @return The execution result, containing for example the net runtime of the program, and the accumulators.
+ *
+ * @throws Exception Thrown, if job submission caused an exception.
+ */
+ JobExecutionResult executeStreamGraph(StreamGraph streamGraph) throws Exception;
--- End diff --
Like PlanExecutor， Exception is still be throwed, which may be mixed with multi exception types.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---