You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2019/11/12 07:58:46 UTC
[flink] 06/24: WIP: cli
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch executors
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7730e5dfc17b34b8ef26d0b3258c1032fa084ec1
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Fri Nov 1 13:38:50 2019 +0100
WIP: cli
---
.../java/org/apache/flink/client/ClientUtils.java | 136 ++++++++++++++++++++
.../org/apache/flink/client/cli/CliFrontend.java | 139 +++------------------
.../flink/client/cli/CliFrontendRunTest.java | 9 +-
3 files changed, 156 insertions(+), 128 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 176d586..d95da8f 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -18,21 +18,32 @@
package org.apache.flink.client;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+import org.apache.flink.client.cli.ExecutionConfigAccessor;
+import org.apache.flink.client.deployment.ClusterClientFactory;
+import org.apache.flink.client.deployment.ClusterClientServiceLoader;
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ContextEnvironment;
import org.apache.flink.client.program.ContextEnvironmentFactory;
import org.apache.flink.client.program.DetachedJobExecutionResult;
import org.apache.flink.client.program.PackagedProgram;
+import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.ProgramMissingJobException;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.ShutdownHookUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,6 +53,7 @@ import java.io.IOException;
import java.net.URISyntaxException;
import java.net.URL;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.jar.JarFile;
@@ -147,6 +159,130 @@ public enum ClientUtils {
}
}
+ public static <ClusterID> void runProgram(
+ final ClusterClientServiceLoader clusterClientServiceLoader,
+ final Configuration configuration,
+ final PackagedProgram program) throws ProgramInvocationException, FlinkException {
+
+ checkNotNull(clusterClientServiceLoader);
+ checkNotNull(configuration);
+ checkNotNull(program);
+
+ final ClusterClientFactory<ClusterID> clusterClientFactory = clusterClientServiceLoader.getClusterClientFactory(configuration);
+ checkNotNull(clusterClientFactory);
+
+ final ClusterDescriptor<ClusterID> clusterDescriptor = clusterClientFactory.createClusterDescriptor(configuration);
+
+ try {
+ final ClusterID clusterId = clusterClientFactory.getClusterId(configuration);
+ final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromConfiguration(configuration);
+ final ClusterClient<ClusterID> client;
+
+ // directly deploy the job if the cluster is started in job mode and detached
+ if (clusterId == null && executionParameters.getDetachedMode()) {
+ int parallelism = executionParameters.getParallelism() == -1 ? 1 : executionParameters.getParallelism();
+
+ final JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, configuration, parallelism);
+
+ final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(configuration);
+ client = clusterDescriptor.deployJobCluster(
+ clusterSpecification,
+ jobGraph,
+ executionParameters.getDetachedMode());
+
+ logAndSysout("Job has been submitted with JobID " + jobGraph.getJobID());
+
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly shut down the client.", e);
+ }
+ } else {
+ final Thread shutdownHook;
+ if (clusterId != null) {
+ client = clusterDescriptor.retrieve(clusterId);
+ shutdownHook = null;
+ } else {
+ // also in job mode we have to deploy a session cluster because the job
+ // might consist of multiple parts (e.g. when using collect)
+ final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(configuration);
+ client = clusterDescriptor.deploySessionCluster(clusterSpecification);
+ // if not running in detached mode, add a shutdown hook to shut down cluster if client exits
+ // there's a race-condition here if cli is killed before shutdown hook is installed
+ if (!executionParameters.getDetachedMode() && executionParameters.isShutdownOnAttachedExit()) {
+ shutdownHook = ShutdownHookUtil.addShutdownHook(client::shutDownCluster, client.getClass().getSimpleName(), LOG);
+ } else {
+ shutdownHook = null;
+ }
+ }
+
+ try {
+ int userParallelism = executionParameters.getParallelism();
+ LOG.debug("User parallelism is set to {}", userParallelism);
+ if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
+ userParallelism = 1;
+ }
+
+ // TODO: 01.11.19 here is where I should simply pass the configuration
+ executeProgram(program, client, userParallelism, executionParameters.getDetachedMode());
+ } finally {
+ if (clusterId == null && !executionParameters.getDetachedMode()) {
+ // terminate the cluster only if we have started it before and if it's not detached
+ try {
+ client.shutDownCluster();
+ } catch (final Exception e) {
+ LOG.info("Could not properly terminate the Flink cluster.", e);
+ }
+ if (shutdownHook != null) {
+ // we do not need the hook anymore as we have just tried to shutdown the cluster.
+ ShutdownHookUtil.removeShutdownHook(shutdownHook, client.getClass().getSimpleName(), LOG);
+ }
+ }
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly shut down the client.", e);
+ }
+ }
+ }
+ } finally {
+ try {
+ clusterDescriptor.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly close the cluster descriptor.", e);
+ }
+ }
+ }
+
+ protected static void executeProgram(
+ PackagedProgram program,
+ ClusterClient<?> client,
+ int parallelism,
+ boolean detached) throws ProgramMissingJobException, ProgramInvocationException {
+ logAndSysout("Starting execution of program");
+
+ JobSubmissionResult result = ClientUtils.executeProgram(client, program, parallelism, detached);
+
+ if (result.isJobExecutionResult()) {
+ logAndSysout("Program execution finished");
+ JobExecutionResult execResult = result.getJobExecutionResult();
+ System.out.println("Job with JobID " + execResult.getJobID() + " has finished.");
+ System.out.println("Job Runtime: " + execResult.getNetRuntime() + " ms");
+ Map<String, Object> accumulatorsResult = execResult.getAllAccumulatorResults();
+ if (accumulatorsResult.size() > 0) {
+ System.out.println("Accumulator Results: ");
+ System.out.println(AccumulatorHelper.getResultsFormatted(accumulatorsResult));
+ }
+ } else {
+ logAndSysout("Job has been submitted with JobID " + result.getJobID());
+ }
+ }
+
+ private static void logAndSysout(String message) {
+ LOG.info(message);
+ System.out.println(message);
+ }
+
public static JobSubmissionResult executeProgram(
ClusterClient<?> client,
PackagedProgram program,
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index da932dd..a1da66d 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -20,17 +20,13 @@ package org.apache.flink.client.cli;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.InvalidProgramException;
-import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.JobSubmissionResult;
-import org.apache.flink.api.common.accumulators.AccumulatorHelper;
import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
import org.apache.flink.client.deployment.ClusterClientFactory;
import org.apache.flink.client.deployment.ClusterClientServiceLoader;
import org.apache.flink.client.deployment.ClusterDescriptor;
-import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.DefaultClusterClientServiceLoader;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.PackagedProgram;
@@ -48,7 +44,6 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.plugin.PluginUtils;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
-import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.security.SecurityConfiguration;
@@ -56,7 +51,6 @@ import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.ShutdownHookUtil;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
@@ -114,8 +108,6 @@ public class CliFrontend {
private final Duration clientTimeout;
- private final int defaultParallelism;
-
private final ClusterClientServiceLoader clusterClientServiceLoader;
public CliFrontend(
@@ -142,7 +134,6 @@ public class CliFrontend {
}
this.clientTimeout = AkkaUtils.getClientTimeout(this.configuration);
- this.defaultParallelism = configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM);
}
// --------------------------------------------------------------------------------------------
@@ -210,103 +201,27 @@ public class CliFrontend {
}
final CustomCommandLine customCommandLine = getActiveCustomCommandLine(commandLine);
- final Configuration executorConfig = customCommandLine.applyCommandLineOptionsToConfiguration(commandLine);
- final Configuration executionConfig = executionParameters.getConfiguration();
+
+ final Configuration effectiveConfig = getEffectiveConfiguration(commandLine, executionParameters, customCommandLine);
try {
- runProgram(executorConfig, executionConfig, program);
+ execute(effectiveConfig, program);
} finally {
program.deleteExtractedLibraries();
}
}
- private <ClusterID> void runProgram(
- Configuration executorConfig,
- Configuration executionConfig,
- PackagedProgram program) throws ProgramInvocationException, FlinkException {
-
- final ClusterClientFactory<ClusterID> clusterClientFactory = clusterClientServiceLoader.getClusterClientFactory(executorConfig);
- checkNotNull(clusterClientFactory);
-
- final ClusterDescriptor<ClusterID> clusterDescriptor = clusterClientFactory.createClusterDescriptor(executorConfig);
-
- try {
- final ClusterID clusterId = clusterClientFactory.getClusterId(executorConfig);
- final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromConfiguration(executionConfig);
- final ClusterClient<ClusterID> client;
-
- // directly deploy the job if the cluster is started in job mode and detached
- if (clusterId == null && executionParameters.getDetachedMode()) {
- int parallelism = executionParameters.getParallelism() == -1 ? defaultParallelism : executionParameters.getParallelism();
-
- final JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, configuration, parallelism);
-
- final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(executorConfig);
- client = clusterDescriptor.deployJobCluster(
- clusterSpecification,
- jobGraph,
- executionParameters.getDetachedMode());
-
- logAndSysout("Job has been submitted with JobID " + jobGraph.getJobID());
-
- try {
- client.close();
- } catch (Exception e) {
- LOG.info("Could not properly shut down the client.", e);
- }
- } else {
- final Thread shutdownHook;
- if (clusterId != null) {
- client = clusterDescriptor.retrieve(clusterId);
- shutdownHook = null;
- } else {
- // also in job mode we have to deploy a session cluster because the job
- // might consist of multiple parts (e.g. when using collect)
- final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(executorConfig);
- client = clusterDescriptor.deploySessionCluster(clusterSpecification);
- // if not running in detached mode, add a shutdown hook to shut down cluster if client exits
- // there's a race-condition here if cli is killed before shutdown hook is installed
- if (!executionParameters.getDetachedMode() && executionParameters.isShutdownOnAttachedExit()) {
- shutdownHook = ShutdownHookUtil.addShutdownHook(client::shutDownCluster, client.getClass().getSimpleName(), LOG);
- } else {
- shutdownHook = null;
- }
- }
-
- try {
- int userParallelism = executionParameters.getParallelism();
- LOG.debug("User parallelism is set to {}", userParallelism);
- if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
- userParallelism = defaultParallelism;
- }
+ protected void execute(final Configuration configuration, final PackagedProgram program) throws FlinkException, ProgramInvocationException {
+ ClientUtils.runProgram(clusterClientServiceLoader, configuration, program);
+ }
- executeProgram(program, client, userParallelism, executionParameters.getDetachedMode());
- } finally {
- if (clusterId == null && !executionParameters.getDetachedMode()) {
- // terminate the cluster only if we have started it before and if it's not detached
- try {
- client.shutDownCluster();
- } catch (final Exception e) {
- LOG.info("Could not properly terminate the Flink cluster.", e);
- }
- if (shutdownHook != null) {
- // we do not need the hook anymore as we have just tried to shutdown the cluster.
- ShutdownHookUtil.removeShutdownHook(shutdownHook, client.getClass().getSimpleName(), LOG);
- }
- }
- try {
- client.close();
- } catch (Exception e) {
- LOG.info("Could not properly shut down the client.", e);
- }
- }
- }
- } finally {
- try {
- clusterDescriptor.close();
- } catch (Exception e) {
- LOG.info("Could not properly close the cluster descriptor.", e);
- }
- }
+ private Configuration getEffectiveConfiguration(CommandLine commandLine, ExecutionConfigAccessor executionParameters, CustomCommandLine customCommandLine) throws FlinkException {
+ // TODO: 01.11.19 all this should be merged nicely, e.g. executionParameters can
+ // take an already existing configuration as a parameter.
+ final Configuration executorConfig = customCommandLine.applyCommandLineOptionsToConfiguration(commandLine);
+ final Configuration executionConfig = executionParameters.getConfiguration();
+ final Configuration effectiveConfig = new Configuration(executorConfig);
+ effectiveConfig.addAll(executionConfig);
+ return effectiveConfig;
}
/**
@@ -342,7 +257,7 @@ public class CliFrontend {
try {
int parallelism = programOptions.getParallelism();
if (ExecutionConfig.PARALLELISM_DEFAULT == parallelism) {
- parallelism = defaultParallelism;
+ parallelism = configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM);
}
LOG.info("Creating program plan dump");
@@ -743,30 +658,6 @@ public class CliFrontend {
// Interaction with programs and JobManager
// --------------------------------------------------------------------------------------------
- protected void executeProgram(
- PackagedProgram program,
- ClusterClient<?> client,
- int parallelism,
- boolean detached) throws ProgramMissingJobException, ProgramInvocationException {
- logAndSysout("Starting execution of program");
-
- JobSubmissionResult result = ClientUtils.executeProgram(client, program, parallelism, detached);
-
- if (result.isJobExecutionResult()) {
- logAndSysout("Program execution finished");
- JobExecutionResult execResult = result.getJobExecutionResult();
- System.out.println("Job with JobID " + execResult.getJobID() + " has finished.");
- System.out.println("Job Runtime: " + execResult.getNetRuntime() + " ms");
- Map<String, Object> accumulatorsResult = execResult.getAllAccumulatorResults();
- if (accumulatorsResult.size() > 0) {
- System.out.println("Accumulator Results: ");
- System.out.println(AccumulatorHelper.getResultsFormatted(accumulatorsResult));
- }
- } else {
- logAndSysout("Job has been submitted with JobID " + result.getJobID());
- }
- }
-
/**
* Creates a Packaged program from the given command line options.
*
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
index d2aff28..72174eb 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
@@ -20,9 +20,10 @@ package org.apache.flink.client.cli;
import org.apache.flink.client.deployment.ClusterClientServiceLoader;
import org.apache.flink.client.deployment.DefaultClusterClientServiceLoader;
-import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.commons.cli.CommandLine;
@@ -198,9 +199,9 @@ public class CliFrontendRunTest extends CliFrontendTestBase {
}
@Override
- protected void executeProgram(PackagedProgram program, ClusterClient client, int parallelism, boolean detached) {
- assertEquals(isDetached, detached);
- assertEquals(expectedParallelism, parallelism);
+ protected void execute(Configuration configuration, PackagedProgram program) {
+ assertEquals(isDetached, !configuration.getBoolean(ExecutionOptions.ATTACHED));
+ assertEquals(expectedParallelism, configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM));
}
}
}