This is an automated email from the ASF dual-hosted git repository.
baiyangtx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 84369c8a8 [AMORO-2652] Support Flink Optimizer run in session cluster
(#2676)
84369c8a8 is described below
commit 84369c8a8d25fa20d2fb36e80c82e75b98ff024e
Author: big face cat <[email protected]>
AuthorDate: Thu Apr 11 19:36:27 2024 +0800
[AMORO-2652] Support Flink Optimizer run in session cluster (#2676)
* [AMORO-2652] Support Flink Optimizer run in session cluster
* fix dependency
* fix doc
* fix doc
* fix comment.
* fix comment.
* shade iceberg
---------
Co-authored-by: huyuanfeng <[email protected]>
Co-authored-by: ZhouJinsong <[email protected]>
---
ams/server/pom.xml | 11 +
.../server/manager/FlinkOptimizerContainer.java | 289 +++++++++++++++++----
.../arctic/server/utils/FlinkClientUtil.java | 48 ++++
docs/admin-guides/managing-optimizers.md | 23 +-
pom.xml | 14 +
5 files changed, 335 insertions(+), 50 deletions(-)
diff --git a/ams/server/pom.xml b/ams/server/pom.xml
index 5cffd07a4..e1092a219 100644
--- a/ams/server/pom.xml
+++ b/ams/server/pom.xml
@@ -267,6 +267,17 @@
<artifactId>hadoop-aws</artifactId>
</dependency>
+ <!-- flink dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime-web</artifactId>
+ </dependency>
+
<!-- runtime dependencies -->
<dependency>
<groupId>org.apache.amoro</groupId>
diff --git
a/ams/server/src/main/java/com/netease/arctic/server/manager/FlinkOptimizerContainer.java
b/ams/server/src/main/java/com/netease/arctic/server/manager/FlinkOptimizerContainer.java
index 84d885d72..c55e4ce00 100644
---
a/ams/server/src/main/java/com/netease/arctic/server/manager/FlinkOptimizerContainer.java
+++
b/ams/server/src/main/java/com/netease/arctic/server/manager/FlinkOptimizerContainer.java
@@ -20,26 +20,56 @@ package com.netease.arctic.server.manager;
import com.netease.arctic.api.OptimizerProperties;
import com.netease.arctic.api.resource.Resource;
+import com.netease.arctic.server.utils.FlinkClientUtil;
import org.apache.commons.lang3.StringUtils;
-import
org.apache.curator.shaded.com.google.common.annotations.VisibleForTesting;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.RestoreMode;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.FileUpload;
+import org.apache.flink.runtime.rest.RestClient;
+import
org.apache.flink.runtime.rest.handler.legacy.messages.ClusterOverviewWithVersion;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.util.RestConstants;
+import org.apache.flink.runtime.webmonitor.handlers.JarListHeaders;
+import org.apache.flink.runtime.webmonitor.handlers.JarListInfo;
+import org.apache.flink.runtime.webmonitor.handlers.JarRunHeaders;
+import org.apache.flink.runtime.webmonitor.handlers.JarRunMessageParameters;
+import org.apache.flink.runtime.webmonitor.handlers.JarRunRequestBody;
+import org.apache.flink.runtime.webmonitor.handlers.JarRunResponseBody;
+import org.apache.flink.runtime.webmonitor.handlers.JarUploadHeaders;
+import org.apache.flink.runtime.webmonitor.handlers.JarUploadResponseBody;
+import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Function;
import org.apache.iceberg.relocated.com.google.common.base.Joiner;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Supplier;
+import org.apache.iceberg.relocated.com.google.common.base.Suppliers;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import
org.apache.iceberg.relocated.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import java.io.BufferedReader;
+import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UncheckedIOException;
+import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -51,6 +81,7 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
public static final String FLINK_CONFIG_PATH = "/conf";
public static final String FLINK_CONFIG_YAML = "/flink-conf.yaml";
public static final String ENV_FLINK_CONF_DIR = "FLINK_CONF_DIR";
+ public static final String FLINK_CLIENT_TIMEOUT_SECOND =
"flink-client-timeout-second";
private static final String DEFAULT_JOB_URI =
"/plugin/optimizer/flink/optimizer-job.jar";
private static final String FLINK_JOB_MAIN_CLASS =
@@ -73,10 +104,12 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
public static final String YARN_APPLICATION_ID_PROPERTY =
"yarn-application-id";
public static final String KUBERNETES_CLUSTER_ID_PROPERTY =
"kubernetes-cluster-id";
+ public static final String SESSION_CLUSTER_JOB_ID = "session-cluster-job-id";
private static final Pattern APPLICATION_ID_PATTERN =
Pattern.compile("(.*)application_(\\d+)_(\\d+)");
private static final int MAX_READ_APP_ID_TIME = 600000; // 10 min
+ private static final long FLINK_CLIENT_TIMEOUT_SECOND_DEFAULT = 30; // 30s
private static final Function<String, String> yarnApplicationIdReader =
readLine -> {
@@ -87,16 +120,28 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
return null;
};
+ private final Supplier<ExecutorService> clientExecutorServiceSupplier =
+ Suppliers.memoize(
+ () ->
+ Executors.newFixedThreadPool(
+ 5,
+ new ThreadFactoryBuilder()
+ .setNameFormat("flink-rest-cluster-client-io-%d")
+ .setDaemon(true)
+ .build()));
+
private Target target;
private String flinkHome;
private String flinkConfDir;
private String jobUri;
+ private long flinkClientTimeoutSecond;
@Override
public void init(String name, Map<String, String> containerProperties) {
super.init(name, containerProperties);
this.flinkHome = getFlinkHome();
this.flinkConfDir = getFlinkConfDir();
+ this.flinkClientTimeoutSecond = getFlinkClientTimeoutSecond();
String runTarget =
Optional.ofNullable(containerProperties.get(FLINK_RUN_TARGET))
@@ -128,30 +173,46 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
@Override
protected Map<String, String> doScaleOut(Resource resource) {
- String startUpArgs = this.buildOptimizerStartupArgsString(resource);
- Runtime runtime = Runtime.getRuntime();
- try {
- String exportCmd = String.join(" && ", exportSystemProperties());
- String startUpCmd = String.format("%s && %s", exportCmd, startUpArgs);
- String[] cmd = {"/bin/sh", "-c", startUpCmd};
- LOG.info("Starting flink optimizer using command : {}", startUpCmd);
- Process exec = runtime.exec(cmd);
- Map<String, String> startUpStatesMap = Maps.newHashMap();
- switch (target) {
- case YARN_PER_JOB:
- case YARN_APPLICATION:
- String applicationId = fetchCommandOutput(exec,
yarnApplicationIdReader);
- if (applicationId != null) {
- startUpStatesMap.put(YARN_APPLICATION_ID_PROPERTY, applicationId);
- }
- break;
- case KUBERNETES_APPLICATION:
- startUpStatesMap.put(KUBERNETES_CLUSTER_ID_PROPERTY,
kubernetesClusterId(resource));
- break;
+ if (target.runByFlinkRestClient()) {
+ try {
+ Configuration configuration =
+ FlinkConf.buildFor(loadFlinkConfig(), getContainerProperties())
+ .withGroupProperties(resource.getProperties())
+ .build()
+ .toConfiguration();
+ JobID jobID = runFlinkOptimizerJob(resource, configuration);
+ Map<String, String> startUpStatesMap = Maps.newHashMap();
+ startUpStatesMap.put(SESSION_CLUSTER_JOB_ID, jobID.toString());
+ return startUpStatesMap;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to scale out flink optimizer in
session cluster.", e);
+ }
+ } else {
+ String startUpArgs = this.buildOptimizerStartupArgsString(resource);
+ Runtime runtime = Runtime.getRuntime();
+ try {
+ String exportCmd = String.join(" && ", exportSystemProperties());
+ String startUpCmd = String.format("%s && %s", exportCmd, startUpArgs);
+ String[] cmd = {"/bin/sh", "-c", startUpCmd};
+ LOG.info("Starting flink optimizer using command : {}", startUpCmd);
+ Process exec = runtime.exec(cmd);
+ Map<String, String> startUpStatesMap = Maps.newHashMap();
+ switch (target) {
+ case YARN_PER_JOB:
+ case YARN_APPLICATION:
+ String applicationId = fetchCommandOutput(exec,
yarnApplicationIdReader);
+ if (applicationId != null) {
+ startUpStatesMap.put(YARN_APPLICATION_ID_PROPERTY,
applicationId);
+ }
+ break;
+ case KUBERNETES_APPLICATION:
+ startUpStatesMap.put(KUBERNETES_CLUSTER_ID_PROPERTY,
kubernetesClusterId(resource));
+ break;
+ }
+ return startUpStatesMap;
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to scale out flink optimizer.",
e);
}
- return startUpStatesMap;
- } catch (IOException e) {
- throw new UncheckedIOException("Failed to scale out flink optimizer.",
e);
}
}
@@ -307,27 +368,56 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
@Override
public void releaseOptimizer(Resource resource) {
- String releaseCommand;
- switch (target) {
- case YARN_APPLICATION:
- case YARN_PER_JOB:
- releaseCommand = buildReleaseYarnCommand(resource);
- break;
- case KUBERNETES_APPLICATION:
- releaseCommand = buildReleaseKubernetesCommand(resource);
- break;
- default:
- throw new IllegalStateException("Unsupported running target: " +
target.getValue());
- }
+ if (target.runByFlinkRestClient()) {
+ Preconditions.checkArgument(
+ resource.getProperties().containsKey(SESSION_CLUSTER_JOB_ID),
+ "Cannot find {} from optimizer start up stats",
+ SESSION_CLUSTER_JOB_ID);
+ String jobId = resource.getProperties().get(SESSION_CLUSTER_JOB_ID);
+ Configuration configuration =
+ FlinkConf.buildFor(loadFlinkConfig(), getContainerProperties())
+ .withGroupProperties(resource.getProperties())
+ .build()
+ .toConfiguration();
+
+ try (RestClusterClient<String> restClusterClient =
+ FlinkClientUtil.getRestClusterClient(configuration)) {
+ Acknowledge ignore =
+ restClusterClient
+ .cancel(JobID.fromHexString(jobId))
+ .get(flinkClientTimeoutSecond, TimeUnit.SECONDS);
+ } catch (TimeoutException timeoutException) {
+ throw new RuntimeException(
+ String.format(
+ "Timed out stopping the optimizer in Flink cluster, "
+ + "please configure a larger timeout via '%s'",
+ FLINK_CLIENT_TIMEOUT_SECOND));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to release flink optimizer", e);
+ }
+ } else {
+ String releaseCommand;
+ switch (target) {
+ case YARN_APPLICATION:
+ case YARN_PER_JOB:
+ releaseCommand = buildReleaseYarnCommand(resource);
+ break;
+ case KUBERNETES_APPLICATION:
+ releaseCommand = buildReleaseKubernetesCommand(resource);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported running target: " +
target.getValue());
+ }
- try {
- String exportCmd = String.join(" && ", exportSystemProperties());
- String releaseCmd = exportCmd + " && " + releaseCommand;
- String[] cmd = {"/bin/sh", "-c", releaseCmd};
- LOG.info("Releasing flink optimizer using command: " + releaseCmd);
- Runtime.getRuntime().exec(cmd);
- } catch (IOException e) {
- throw new UncheckedIOException("Failed to release flink optimizer.", e);
+ try {
+ String exportCmd = String.join(" && ", exportSystemProperties());
+ String releaseCmd = exportCmd + " && " + releaseCommand;
+ String[] cmd = {"/bin/sh", "-c", releaseCmd};
+ LOG.info("Releasing flink optimizer using command: " + releaseCmd);
+ Runtime.getRuntime().exec(cmd);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to release flink optimizer.",
e);
+ }
}
}
@@ -371,6 +461,89 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
this.flinkHome, options);
}
+ /** Submit the OptimizerJob to the flink cluster through Flink RestClient. */
+ protected JobID runFlinkOptimizerJob(Resource resource, Configuration
configuration)
+ throws Exception {
+ try (RestClusterClient<String> client =
FlinkClientUtil.getRestClusterClient(configuration)) {
+ ClusterOverviewWithVersion overviewWithVersion =
+ client.getClusterOverview().get(flinkClientTimeoutSecond,
TimeUnit.SECONDS);
+ LOG.debug("cluster overview: {}", overviewWithVersion);
+ File jarFile = new File(jobUri);
+ Preconditions.checkArgument(
+ jarFile.exists(), String.format("The jar file %s not exists",
jarFile.getAbsolutePath()));
+
+ URL url = new URL(client.getWebInterfaceURL());
+ return runJar(
+ uploadJar(jarFile, url.getHost(), url.getPort(), configuration),
configuration, resource);
+ }
+ }
+
+ @VisibleForTesting
+ protected String uploadJar(File jarFile, String host, int port,
Configuration configuration)
+ throws Exception {
+ Preconditions.checkArgument(
+ jarFile.exists(), String.format("The jar file %s not exists",
jarFile.getAbsolutePath()));
+ try (RestClient restClient =
+ FlinkClientUtil.getRestClient(configuration,
clientExecutorServiceSupplier.get())) {
+ // First check whether the jar already exists in the cluster. If it
exists, there is no need
+ // to upload it multiple times. If it does not exist, upload and return
jarId.
+ JarListHeaders jarListHeaders = JarListHeaders.getInstance();
+ JarListInfo jarListInfo =
+ restClient
+ .sendRequest(
+ host,
+ port,
+ jarListHeaders,
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance())
+ .get(flinkClientTimeoutSecond, TimeUnit.SECONDS);
+ Optional<JarListInfo.JarFileInfo> optimizerJarOptional =
+ jarListInfo.jarFileList.stream()
+ .filter(jarFileInfo ->
jarFile.getName().equals(jarFileInfo.name))
+ .findFirst();
+ if (optimizerJarOptional.isPresent()) {
+ return optimizerJarOptional.get().id;
+ }
+ JarUploadHeaders jarUploadHeaders = JarUploadHeaders.getInstance();
+ JarUploadResponseBody jarUploadResponseBody =
+ restClient
+ .sendRequest(
+ host,
+ port,
+ jarUploadHeaders,
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance(),
+ Collections.singletonList(
+ new FileUpload(jarFile.toPath(),
RestConstants.CONTENT_TYPE_JAR)))
+ .get(flinkClientTimeoutSecond, TimeUnit.SECONDS);
+ return jarUploadResponseBody
+ .getFilename()
+ .substring(jarUploadResponseBody.getFilename().lastIndexOf("/") + 1);
+ }
+ }
+
+ @VisibleForTesting
+ protected JobID runJar(String jarId, Configuration configuration, Resource
resource)
+ throws Exception {
+ JarRunHeaders headers = JarRunHeaders.getInstance();
+ JarRunMessageParameters parameters =
headers.getUnresolvedMessageParameters();
+ parameters.jarIdPathParameter.resolve(jarId);
+ String args = super.buildOptimizerStartupArgsString(resource);
+ JobID jobID = JobID.generate();
+ JarRunRequestBody runRequestBody =
+ new JarRunRequestBody(
+ FLINK_JOB_MAIN_CLASS, args, null, null, jobID, true, null,
RestoreMode.DEFAULT, null);
+ LOG.info("Submitting job: {} to session cluster, args: {}", jobID, args);
+ try (RestClusterClient<String> restClusterClient =
+ FlinkClientUtil.getRestClusterClient(configuration)) {
+ JarRunResponseBody jarRunResponseBody =
+ restClusterClient
+ .sendRequest(headers, parameters, runRequestBody)
+ .get(flinkClientTimeoutSecond, TimeUnit.SECONDS);
+ return jarRunResponseBody.getJobId();
+ }
+ }
+
private String getFlinkHome() {
String flinkHome = getContainerProperties().get(FLINK_HOME_PROPERTY);
Preconditions.checkNotNull(
@@ -388,21 +561,30 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
return this.flinkHome + FLINK_CONFIG_PATH;
}
+ private long getFlinkClientTimeoutSecond() {
+ return getContainerProperties().get(FLINK_CLIENT_TIMEOUT_SECOND) != null
+ ?
Long.parseLong(getContainerProperties().get(FLINK_CLIENT_TIMEOUT_SECOND))
+ : FLINK_CLIENT_TIMEOUT_SECOND_DEFAULT;
+ }
+
private String kubernetesClusterId(Resource resource) {
return "amoro-optimizer-" + resource.getResourceId();
}
private enum Target {
- YARN_PER_JOB("yarn-per-job", false),
- YARN_APPLICATION("yarn-application", true),
- KUBERNETES_APPLICATION("kubernetes-application", true);
+ YARN_PER_JOB("yarn-per-job", false, false),
+ YARN_APPLICATION("yarn-application", true, false),
+ KUBERNETES_APPLICATION("kubernetes-application", true, false),
+ SESSION("session", false, true);
private final String value;
private final boolean applicationMode;
+ private final boolean runByFlinkRestClient;
- Target(String value, boolean applicationMode) {
+ Target(String value, boolean applicationMode, boolean
runByFlinkRestClient) {
this.value = value;
this.applicationMode = applicationMode;
+ this.runByFlinkRestClient = runByFlinkRestClient;
}
public static Target valueToEnum(String value) {
@@ -416,6 +598,10 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
return applicationMode;
}
+ public boolean runByFlinkRestClient() {
+ return runByFlinkRestClient;
+ }
+
public String getValue() {
return value;
}
@@ -476,6 +662,13 @@ public class FlinkOptimizerContainer extends
AbstractResourceContainer {
return new Builder(flinkConf, containerProperties);
}
+ public Configuration toConfiguration() {
+ HashMap<String, String> config = new HashMap<>();
+ config.putAll(flinkConf);
+ config.putAll(flinkOptions);
+ return Configuration.fromMap(config);
+ }
+
public static class Builder {
final Map<String, String> flinkConf;
diff --git
a/ams/server/src/main/java/com/netease/arctic/server/utils/FlinkClientUtil.java
b/ams/server/src/main/java/com/netease/arctic/server/utils/FlinkClientUtil.java
new file mode 100644
index 000000000..814c6ebee
--- /dev/null
+++
b/ams/server/src/main/java/com/netease/arctic/server/utils/FlinkClientUtil.java
@@ -0,0 +1,48 @@
+/*
+ * 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 com.netease.arctic.server.utils;
+
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.RestClient;
+
+import java.util.concurrent.ExecutorService;
+
+public class FlinkClientUtil {
+
+ /**
+ * Get Flink RestClusterClient.
+ *
+ * @param configuration {@link org.apache.flink.configuration.RestOptions}
+ */
+ public static RestClusterClient<String> getRestClusterClient(Configuration
configuration)
+ throws Exception {
+ return new RestClusterClient<>(configuration,
"Amoro-optimizer-session-cluster");
+ }
+
+ /**
+ * Get Flink RestClient.
+ *
+ * @param configuration {@link org.apache.flink.configuration.RestOptions}
+ */
+ public static RestClient getRestClient(Configuration configuration,
ExecutorService service)
+ throws Exception {
+ return new RestClient(configuration, service);
+ }
+}
diff --git a/docs/admin-guides/managing-optimizers.md
b/docs/admin-guides/managing-optimizers.md
index 25dd8c1b9..bf9d19ad8 100644
--- a/docs/admin-guides/managing-optimizers.md
+++ b/docs/admin-guides/managing-optimizers.md
@@ -40,7 +40,7 @@ FlinkOptimizerContainer support the following properties:
| Property Name | Required | Default Value | Description
|
|---------------------------|----------|---------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| flink-home | true | N/A | Flink installation
location
|
-| target | true | yarn-per-job | flink job deployed
target, available values `yarn-per-job`, `yarn-application`,
`kubernetes-application`
|
+| target | true | yarn-per-job | flink job deployed
target, available values `yarn-per-job`, `yarn-application`,
`kubernetes-application`, `session`
|
| job-uri | false | N/A | The jar uri of flink
optimizer job. This is required if target is application mode.
|
| ams-optimizing-uri | false | N/A | uri of AMS thrift
self-optimizing endpoint. This could be used if the ams.server-expose-host is
not available
|
| export.\<key\> | false | N/A | environment variables
will be exported during job submit
|
@@ -48,7 +48,7 @@ FlinkOptimizerContainer support the following properties:
| export.HADOOP_CONF_DIR | false | N/A | Direction which holds
the configuration files for the hadoop cluster (including hdfs-site.xml,
core-site.xml, yarn-site.xml ). If the hadoop cluster has kerberos
authentication enabled, you need to prepare an additional krb5.conf and a
keytab file for the user to submit tasks |
| export.JVM_ARGS | false | N/A | you can configure flink
to run additional configuration parameters, here is an example of configuring
krb5.conf, specify the address of krb5.conf to be used by Flink when committing
via `-Djava.security.krb5.conf=/opt/krb5.conf`
|
| export.HADOOP_USER_NAME | false | N/A | the username used to
submit tasks to yarn, used for simple authentication
|
-| export.FLINK_CONF_DIR | false | N/A | the directory where
flink_conf.yaml is located
|
+| export.FLINK_CONF_DIR | false | N/A | the directory where
flink_conf.yaml is located
|
| flink-conf.\<key\> | false | N/A | [Flink Configuration
Options](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/)
will be passed to cli by `-Dkey=value`,
|
{{< hint info >}}
@@ -89,6 +89,25 @@ containers:
flink-conf.kubernetes.service-account: flink
# Service account that is used within kubernetes cluster.
```
+An example for flink session mode:
+
+```yaml
+containers:
+ - name: flinkContainer
+ container-impl: com.netease.arctic.server.manager.FlinkOptimizerContainer
+ properties:
+ target: session
# Flink run in session cluster
+ job-uri: "local:///opt/flink/usrlib/optimizer-job.jar"
# Optimizer job main jar
+ ams-optimizing-uri: thrift://ams.amoro.service.local:1261
# AMS optimizing uri
+ export.FLINK_CONF_DIR: /opt/flink/conf/
# Flink config dir, flink-conf.yaml should e in this dir, contains the
rest connection parameters of the session cluster
+ flink-conf.high-availability: zookeeper
# Flink high availability mode, reference:
https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/config/#high-availability
+ flink-conf.high-availability.zookeeper.quorum: xxx:2181
+ flink-conf.high-availability.zookeeper.path.root: /flink
+ flink-conf.high-availability.cluster-id: amoro-optimizer-cluster
+ flink-conf.high-availability.storageDir: hdfs://xxx/xxx/xxx
+ flink-conf.rest.address: localhost:8081
# If the session cluster is not high availability mode, please configure
the restaddress of jobmanager
+```
+
### Spark container
Spark container is another way to start Optimizer through Spark jobs. With
Spark, you can easily deploy Optimizer
diff --git a/pom.xml b/pom.xml
index 860c6915d..d2e316499 100644
--- a/pom.xml
+++ b/pom.xml
@@ -128,6 +128,7 @@
<lucene.version>8.11.2</lucene.version>
<bitmap.version>1.0.1</bitmap.version>
<prometheus.version>0.16.0</prometheus.version>
+ <flink.version>1.18.1</flink.version>
</properties>
<dependencies>
@@ -768,6 +769,19 @@
<version>${lucene.version}</version>
</dependency>
+ <!-- flink dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime-web</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
