forked from apache/flink
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[FLINK-4930] [client, yarn] Implement FLIP-6 YARN client
Summary: Implement FLIP-6 YARN client Test Plan: NA Reviewers: biao.liub Differential Revision: http:https://phabricator.taobao.net/D6563
- Loading branch information
1 parent
2a7dbda
commit 3695a8e
Showing
5 changed files
with
524 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
169 changes: 169 additions & 0 deletions
169
flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClientV2.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,169 @@ | ||
| /* | ||
| * 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:https://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.yarn; | ||
|
|
||
| import org.apache.flink.api.common.JobSubmissionResult; | ||
| import org.apache.flink.client.program.ClusterClient; | ||
| import org.apache.flink.client.program.ProgramInvocationException; | ||
| import org.apache.flink.runtime.clusterframework.ApplicationStatus; | ||
| import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse; | ||
| import org.apache.flink.runtime.jobgraph.JobGraph; | ||
| import org.apache.hadoop.yarn.api.records.ApplicationId; | ||
| import org.apache.hadoop.yarn.api.records.ApplicationReport; | ||
| import org.apache.hadoop.yarn.api.records.YarnApplicationState; | ||
| import org.apache.hadoop.yarn.client.api.YarnClient; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.InetSocketAddress; | ||
| import java.net.URL; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Java representation of a running Flink job on YARN. | ||
| * Since flip-6, a flink job will be run as a yarn job by default, each job has a jobmaster, | ||
| * so this class will be used as a client to communicate with yarn and start the job on yarn. | ||
| */ | ||
| public class YarnClusterClientV2 extends ClusterClient { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(YarnClusterClientV2.class); | ||
|
|
||
| private YarnClient yarnClient; | ||
|
|
||
| private final AbstractYarnClusterDescriptor clusterDescriptor; | ||
|
|
||
| private ApplicationId appId; | ||
|
|
||
| private String trackingURL; | ||
|
|
||
| /** | ||
| * Create a client to communicate with YARN cluster. | ||
| * | ||
| * @param clusterDescriptor The descriptor used to create yarn job | ||
| * @param flinkConfig Flink configuration | ||
| * @throws java.io.IOException | ||
| */ | ||
| public YarnClusterClientV2( | ||
| final AbstractYarnClusterDescriptor clusterDescriptor, | ||
| org.apache.flink.configuration.Configuration flinkConfig) throws IOException { | ||
|
|
||
| super(flinkConfig); | ||
|
|
||
| this.clusterDescriptor = clusterDescriptor; | ||
| this.yarnClient = clusterDescriptor.getYarnClient(); | ||
| this.trackingURL = ""; | ||
| } | ||
|
|
||
| @Override | ||
| public org.apache.flink.configuration.Configuration getFlinkConfiguration() { | ||
| return flinkConfig; | ||
| } | ||
|
|
||
| @Override | ||
| public int getMaxSlots() { | ||
| // Now need not set max slot | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean hasUserJarsInClassPath(List<URL> userJarFiles) { | ||
| return clusterDescriptor.hasUserJarFiles(userJarFiles); | ||
| } | ||
|
|
||
| @Override | ||
| protected JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException { | ||
| try { | ||
| // Create application via yarnClient | ||
| ApplicationReport report = this.clusterDescriptor.startAppMaster(jobGraph, yarnClient); | ||
| if (report.getYarnApplicationState().equals(YarnApplicationState.RUNNING)) { | ||
| appId = report.getApplicationId(); | ||
| trackingURL = report.getTrackingUrl(); | ||
| logAndSysout("Please refer to " + getWebInterfaceURL() | ||
| + " for the running status of job " + jobGraph.getJobID().toString()); | ||
| //TODO: not support attach mode now | ||
| return new JobSubmissionResult(jobGraph.getJobID()); | ||
| } | ||
| else { | ||
| throw new ProgramInvocationException("Fail to submit the job."); | ||
| } | ||
| } | ||
| catch (Exception e) { | ||
| throw new ProgramInvocationException("Fail to submit the job", e.getCause()); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String getWebInterfaceURL() { | ||
| // there seems to be a difference between HD 2.2.0 and 2.6.0 | ||
| if(!trackingURL.startsWith("http:https://")) { | ||
| return "http:https://" + trackingURL; | ||
| } else { | ||
| return trackingURL; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String getClusterIdentifier() { | ||
| return "Yarn cluster with application id " + getApplicationId(); | ||
| } | ||
|
|
||
| /** | ||
| * This method is only available if the cluster hasn't been started in detached mode. | ||
| */ | ||
| @Override | ||
| public GetClusterStatusResponse getClusterStatus() { | ||
| throw new UnsupportedOperationException("Not support getClusterStatus since Flip-6."); | ||
| } | ||
|
|
||
| public ApplicationStatus getApplicationStatus() { | ||
| //TODO: this method is useful for later | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public List<String> getNewMessages() { | ||
| throw new UnsupportedOperationException("Not support getNewMessages since Flip-6."); | ||
| } | ||
|
|
||
| @Override | ||
| public void finalizeCluster() { | ||
| throw new UnsupportedOperationException("Not support finalizeCluster since Flip-6."); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isDetached() { | ||
| return super.isDetached() || clusterDescriptor.isDetachedMode(); | ||
| } | ||
|
|
||
| @Override | ||
| public void waitForClusterToBeReady() { | ||
| throw new UnsupportedOperationException("Not support waitForClusterToBeReady since Flip-6."); | ||
| } | ||
|
|
||
| @Override | ||
| public InetSocketAddress getJobManagerAddress() { | ||
| //TODO: just return a local address in order to be compatible with createClient in CliFrontend | ||
| return new InetSocketAddress("localhost", 0); | ||
| } | ||
|
|
||
| public ApplicationId getApplicationId() { | ||
| return appId; | ||
| } | ||
|
|
||
| } |
34 changes: 34 additions & 0 deletions
34
flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptorV2.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| /* | ||
| * 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:https://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.yarn; | ||
|
|
||
|
|
||
| /** | ||
| * Implementation of {@link org.apache.flink.yarn.AbstractYarnClusterDescriptor} which is used to start the new application master for a job under flip-6. | ||
| * This implementation is now however tricky, since YarnClusterDescriptorV2 is related YarnClusterClientV2, but AbstractYarnClusterDescriptor is related | ||
| * to YarnClusterClient. We should let YarnClusterDescriptorV2 implements ClusterDescriptor<YarnClusterClientV2>. | ||
| * However, in order to use the code in AbstractYarnClusterDescriptor for setting environments and so on, we make YarnClusterDescriptorV2 as now. | ||
| */ | ||
| public class YarnClusterDescriptorV2 extends AbstractYarnClusterDescriptor { | ||
|
|
||
| @Override | ||
| protected Class<?> getApplicationMasterClass() { | ||
| return YarnFlinkApplicationMasterRunner.class; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.