| /** |
| * 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.tez.client; |
| |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.net.UnknownHostException; |
| import java.text.NumberFormat; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.TimeUnit; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.Objects; |
| |
| import javax.annotation.Nullable; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.CommonConfigurationKeys; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.tez.common.JavaOptsChecker; |
| import org.apache.tez.common.TezCommonUtils; |
| import org.apache.tez.common.counters.Limits; |
| import org.apache.tez.dag.api.TezConfigurationConstants; |
| import org.apache.tez.serviceplugins.api.ServicePluginsDescriptor; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.hadoop.classification.InterfaceAudience.Private; |
| import org.apache.hadoop.classification.InterfaceAudience.Public; |
| import org.apache.hadoop.classification.InterfaceStability.Evolving; |
| import org.apache.hadoop.classification.InterfaceStability.Unstable; |
| import org.apache.hadoop.security.Credentials; |
| import org.apache.hadoop.yarn.api.records.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.ApplicationReport; |
| import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; |
| import org.apache.hadoop.yarn.api.records.LocalResource; |
| import org.apache.hadoop.yarn.api.records.LocalResourceType; |
| import org.apache.hadoop.yarn.api.records.YarnApplicationState; |
| import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException; |
| import org.apache.hadoop.yarn.exceptions.YarnException; |
| import org.apache.hadoop.util.Time; |
| import org.apache.tez.common.ReflectionUtils; |
| import org.apache.tez.common.security.JobTokenSecretManager; |
| import org.apache.tez.dag.api.DAG; |
| import org.apache.tez.dag.api.DAGSubmissionTimedOut; |
| import org.apache.tez.dag.api.DagTypeConverters; |
| import org.apache.tez.dag.api.HistoryLogLevel; |
| import org.apache.tez.dag.api.PreWarmVertex; |
| import org.apache.tez.dag.api.SessionNotReady; |
| import org.apache.tez.dag.api.SessionNotRunning; |
| import org.apache.tez.dag.api.TezConfiguration; |
| import org.apache.tez.dag.api.TezConstants; |
| import org.apache.tez.dag.api.TezException; |
| import org.apache.tez.dag.api.client.DAGClient; |
| import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolBlockingPB; |
| import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolRPC.GetAMStatusRequestProto; |
| import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolRPC.SubmitDAGRequestProto; |
| import org.apache.tez.dag.api.records.DAGProtos.DAGPlan; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import org.apache.tez.common.Preconditions; |
| import com.google.common.collect.Maps; |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import com.google.protobuf.ServiceException; |
| |
| /** |
| * TezClient is used to submit Tez DAGs for execution. DAG's are executed via a |
| * Tez App Master. TezClient can run the App Master in session or non-session |
| * mode. <br> |
| * In non-session mode, each DAG is executed in a different App Master that |
| * exits after the DAG execution completes. <br> |
| * In session mode, the TezClient creates a single instance of the App Master |
| * and all DAG's are submitted to the same App Master.<br> |
| * Session mode may give better performance when a series of DAGs need to |
| * executed because it enables resource re-use across those DAGs. Non-session |
| * mode should be used when the user wants to submit a single DAG or wants to |
| * disconnect from the cluster after submitting a set of unrelated DAGs. <br> |
| * If API recommendations are followed, then the choice of running in session or |
| * non-session mode is transparent to writing the application. By changing the |
| * session mode configuration, the same application can be running in session or |
| * non-session mode. |
| */ |
| @Public |
| public class TezClient { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(TezClient.class); |
| |
| private static final String appIdStrPrefix = "application"; |
| private static final String APPLICATION_ID_PREFIX = appIdStrPrefix + '_'; |
| private static final long PREWARM_WAIT_MS = 500; |
| |
| @VisibleForTesting |
| static final String NO_CLUSTER_DIAGNOSTICS_MSG = "No cluster diagnostics found."; |
| |
| @VisibleForTesting |
| final String clientName; |
| private ApplicationId sessionAppId; |
| private ApplicationId lastSubmittedAppId; |
| @VisibleForTesting |
| final AMConfiguration amConfig; |
| @VisibleForTesting |
| FrameworkClient frameworkClient; |
| private String diagnostics; |
| @VisibleForTesting |
| final boolean isSession; |
| private final AtomicBoolean sessionStarted = new AtomicBoolean(false); |
| private final AtomicBoolean sessionStopped = new AtomicBoolean(false); |
| /** Tokens which will be required for all DAGs submitted to this session. */ |
| private Credentials sessionCredentials = new Credentials(); |
| private long clientTimeout; |
| Map<String, LocalResource> cachedTezJarResources; |
| boolean usingTezArchiveDeploy = false; |
| private static final long SLEEP_FOR_READY = 500; |
| private JobTokenSecretManager jobTokenSecretManager = |
| new JobTokenSecretManager(); |
| private final Map<String, LocalResource> additionalLocalResources = Maps.newHashMap(); |
| @VisibleForTesting |
| final TezApiVersionInfo apiVersionInfo; |
| @VisibleForTesting |
| final ServicePluginsDescriptor servicePluginsDescriptor; |
| private JavaOptsChecker javaOptsChecker = null; |
| private DAGClient prewarmDagClient = null; |
| private int preWarmDAGCounter = 0; |
| |
| /* max submitDAG request size through IPC; beyond this we transfer them in the same way we transfer local resource */ |
| private int maxSubmitDAGRequestSizeThroughIPC; |
| /* this counter counts number of serialized DAGPlan and is used to give unique name to each serialized DAGPlan */ |
| private AtomicInteger serializedSubmitDAGPlanRequestCounter = new AtomicInteger(0); |
| private FileSystem stagingFs = null; |
| |
| private ScheduledExecutorService amKeepAliveService; |
| |
| private final Map<String, UserGroupInformation> ugiMap; |
| |
| private TezClient(String name, TezConfiguration tezConf) { |
| this(name, tezConf, tezConf.getBoolean( |
| TezConfiguration.TEZ_AM_SESSION_MODE, TezConfiguration.TEZ_AM_SESSION_MODE_DEFAULT)); |
| } |
| |
| @Private |
| TezClient(String name, TezConfiguration tezConf, |
| @Nullable Map<String, LocalResource> localResources, |
| @Nullable Credentials credentials) { |
| this(name, tezConf, tezConf.getBoolean( |
| TezConfiguration.TEZ_AM_SESSION_MODE, TezConfiguration.TEZ_AM_SESSION_MODE_DEFAULT), |
| localResources, credentials); |
| } |
| |
| private TezClient(String name, TezConfiguration tezConf, boolean isSession) { |
| this(name, tezConf, isSession, null, null); |
| } |
| |
| @Private |
| protected TezClient(String name, TezConfiguration tezConf, boolean isSession, |
| @Nullable Map<String, LocalResource> localResources, |
| @Nullable Credentials credentials) { |
| this(name, tezConf, isSession, localResources, credentials, null); |
| } |
| |
| @Private |
| protected TezClient(String name, TezConfiguration tezConf, boolean isSession, |
| @Nullable Map<String, LocalResource> localResources, |
| @Nullable Credentials credentials, ServicePluginsDescriptor servicePluginsDescriptor) { |
| this.clientName = name; |
| this.isSession = isSession; |
| // Set in conf for local mode AM to figure out whether in session mode or not |
| tezConf.setBoolean(TezConfiguration.TEZ_AM_SESSION_MODE, isSession); |
| try { |
| InetAddress ip = InetAddress.getLocalHost(); |
| if (ip != null) { |
| tezConf.set(TezConfigurationConstants.TEZ_SUBMIT_HOST, ip.getCanonicalHostName()); |
| tezConf.set(TezConfigurationConstants.TEZ_SUBMIT_HOST_ADDRESS, ip.getHostAddress()); |
| } |
| } catch (UnknownHostException e) { |
| LOG.warn("The host name of the client the tez application was submitted from was unable to be retrieved", e); |
| } |
| |
| this.ugiMap = new HashMap<>(); |
| this.amConfig = new AMConfiguration(tezConf, localResources, credentials); |
| this.apiVersionInfo = new TezApiVersionInfo(); |
| this.servicePluginsDescriptor = servicePluginsDescriptor; |
| this.maxSubmitDAGRequestSizeThroughIPC = tezConf.getInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, |
| CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT) - |
| tezConf.getInt(TezConfiguration.TEZ_IPC_PAYLOAD_RESERVED_BYTES, |
| TezConfiguration.TEZ_IPC_PAYLOAD_RESERVED_BYTES_DEFAULT); |
| Limits.setConfiguration(tezConf); |
| |
| LOG.info("Tez Client Version: " + apiVersionInfo.toString()); |
| } |
| |
| |
| /** |
| * Create a new TezClientBuilder. This can be used to setup additional parameters |
| * like session mode, local resources, credentials, servicePlugins, etc. |
| * <p/> |
| * If session mode is not specified in the builder, this will be inferred from |
| * the provided TezConfiguration. |
| * |
| * @param name Name of the client. Used for logging etc. This will also be used |
| * as app master name is session mode |
| * @param tezConf Configuration for the framework |
| * @return An instance of {@link org.apache.tez.client.TezClient.TezClientBuilder} |
| * which can be used to construct the final TezClient. |
| */ |
| public static TezClientBuilder newBuilder(String name, TezConfiguration tezConf) { |
| return new TezClientBuilder(name, tezConf); |
| } |
| |
| /** |
| * Create a new TezClient. Session or non-session execution mode will be |
| * inferred from configuration. |
| * @param name |
| * Name of the client. Used for logging etc. This will also be used |
| * as app master name is session mode |
| * @param tezConf |
| * Configuration for the framework |
| */ |
| public static TezClient create(String name, TezConfiguration tezConf) { |
| return new TezClient(name, tezConf); |
| } |
| |
| /** |
| * Create a new TezClient. Session or non-session execution mode will be |
| * inferred from configuration. Set the initial resources and security |
| * credentials for the App Master. If app master resources/credentials are |
| * needed then this is the recommended method for session mode execution. |
| * |
| * @param name |
| * Name of the client. Used for logging etc. This will also be used |
| * as app master name is session mode |
| * @param tezConf |
| * Configuration for the framework |
| * @param localFiles |
| * local files for the App Master |
| * @param credentials |
| * Set security credentials to be used inside the app master, if |
| * needed. Tez App Master needs credentials to access the staging |
| * directory and for most HDFS cases these are automatically obtained |
| * by Tez client. If the staging directory is on a file system for |
| * which credentials cannot be obtained or for any credentials needed |
| * by user code running inside the App Master, credentials must be |
| * supplied by the user. These will be used by the App Master for the |
| * next DAG. <br> |
| * In session mode, credentials, if needed, must be set before |
| * calling start() |
| */ |
| public static TezClient create(String name, TezConfiguration tezConf, |
| @Nullable Map<String, LocalResource> localFiles, |
| @Nullable Credentials credentials) { |
| return new TezClient(name, tezConf, localFiles, credentials); |
| } |
| |
| /** |
| * Create a new TezClient with AM session mode set explicitly. This overrides |
| * the setting from configuration. |
| * @param name |
| * Name of the client. Used for logging etc. This will also be used |
| * as app master name is session mode |
| * @param tezConf Configuration for the framework |
| * @param isSession The AM will run in session mode or not |
| */ |
| public static TezClient create(String name, TezConfiguration tezConf, boolean isSession) { |
| return new TezClient(name, tezConf, isSession); |
| } |
| |
| /** |
| * Create a new TezClient with AM session mode set explicitly. This overrides |
| * the setting from configuration. |
| * Set the initial files and security credentials for the App Master. |
| * @param name |
| * Name of the client. Used for logging etc. This will also be used |
| * as app master name is session mode |
| * @param tezConf Configuration for the framework |
| * @param isSession The AM will run in session mode or not |
| * @param localFiles local files for the App Master |
| * @param credentials credentials for the App Master |
| */ |
| public static TezClient create(String name, TezConfiguration tezConf, boolean isSession, |
| @Nullable Map<String, LocalResource> localFiles, |
| @Nullable Credentials credentials) { |
| return new TezClient(name, tezConf, isSession, localFiles, credentials); |
| } |
| |
| /** |
| * Add local files for the DAG App Master. These may be files, archives, |
| * jars etc.<br> |
| * <p> |
| * In non-session mode these will be added to the files of the App Master |
| * to be launched for the next DAG. Files added via this method will |
| * accumulate and be used for every new App Master until |
| * {@link #clearAppMasterLocalFiles()} is invoked. <br> |
| * <p> |
| * In session mode, the recommended usage is to add all files before |
| * calling start() so that all needed files are available to the app |
| * master before it starts. When called after start(), these local files |
| * will be re-localized to the running session DAG App Master and will be |
| * added to its classpath for execution of this DAG. |
| * <p> |
| * Caveats for invoking this method after start() in session mode: files |
| * accumulate across DAG submissions and are never removed from the classpath. |
| * Only LocalResourceType.FILE is supported. All files will be treated as |
| * private. |
| * |
| * @param localFiles the files to be made available in the AM |
| */ |
| public synchronized void addAppMasterLocalFiles(Map<String, LocalResource> localFiles) { |
| Objects.requireNonNull(localFiles); |
| if (isSession && sessionStarted.get()) { |
| additionalLocalResources.putAll(localFiles); |
| } |
| amConfig.addAMLocalResources(localFiles); |
| } |
| |
| /** |
| * If the next DAG App Master needs different local files, then use this |
| * method to clear the local files and then add the new local files |
| * using {@link #addAppMasterLocalFiles(Map)}. This method is a no-op in session mode, |
| * after start() is called. |
| */ |
| public synchronized void clearAppMasterLocalFiles() { |
| amConfig.clearAMLocalResources(); |
| } |
| |
| /** |
| * Set security credentials to be used inside the app master, if needed. Tez App |
| * Master needs credentials to access the staging directory and for most HDFS |
| * cases these are automatically obtained by Tez client. If the staging |
| * directory is on a file system for which credentials cannot be obtained or |
| * for any credentials needed by user code running inside the App Master, |
| * credentials must be supplied by the user. These will be used by the App |
| * Master for the next DAG. <br>In session mode, credentials, if needed, must be |
| * set before calling start() |
| * |
| * @param credentials credentials |
| */ |
| public synchronized void setAppMasterCredentials(Credentials credentials) { |
| Preconditions |
| .checkState(!sessionStarted.get(), |
| "Credentials cannot be set after the session App Master has been started"); |
| amConfig.setCredentials(credentials); |
| } |
| |
| /** |
| * Sets the history log level for this session. It will be in effect for DAGs submitted after this |
| * call. |
| * |
| * @param historyLogLevel The log level to be used. |
| */ |
| public synchronized void setHistoryLogLevel(HistoryLogLevel historyLogLevel) { |
| amConfig.getTezConfiguration().setEnum(TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL, |
| historyLogLevel); |
| } |
| |
| /** |
| * Start the client. This establishes a connection to the YARN cluster. |
| * In session mode, this start the App Master thats runs all the DAGs in the |
| * session. |
| * @throws TezException |
| * @throws IOException |
| */ |
| public synchronized void start() throws TezException, IOException { |
| startFrameworkClient(); |
| setupJavaOptsChecker(); |
| |
| if (isSession) { |
| LOG.info("Session mode. Starting session."); |
| TezClientUtils.processTezLocalCredentialsFile(sessionCredentials, |
| amConfig.getTezConfiguration()); |
| |
| clientTimeout = amConfig.getTezConfiguration().getInt( |
| TezConfiguration.TEZ_SESSION_CLIENT_TIMEOUT_SECS, |
| TezConfiguration.TEZ_SESSION_CLIENT_TIMEOUT_SECS_DEFAULT); |
| |
| try { |
| if (sessionAppId == null) { |
| sessionAppId = createApplication(); |
| } |
| |
| ApplicationSubmissionContext appContext = setupApplicationContext(); |
| frameworkClient.submitApplication(appContext); |
| ApplicationReport appReport = frameworkClient.getApplicationReport(sessionAppId); |
| LOG.info("The url to track the Tez Session: " + appReport.getTrackingUrl()); |
| sessionStarted.set(true); |
| } catch (YarnException e) { |
| cleanStagingDir(); |
| throw new TezException(e); |
| } catch (IOException e) { |
| cleanStagingDir(); |
| throw new TezException(e); |
| } |
| |
| startClientHeartbeat(); |
| this.stagingFs = FileSystem.get(amConfig.getTezConfiguration()); |
| } |
| } |
| |
| private void cleanStagingDir() { |
| Configuration conf = amConfig.getTezConfiguration(); |
| String appId = sessionAppId.toString(); |
| Path stagingDir = TezCommonUtils.getTezSystemStagingPath(conf, appId); |
| boolean isStgDeleted = false; |
| try { |
| FileSystem fs = stagingDir.getFileSystem(conf); |
| isStgDeleted = fs.delete(stagingDir, true); |
| } catch (IOException ioe) { |
| LOG.error("Error deleting staging dir " + stagingDir, ioe); |
| } finally { |
| LOG.info("Staging dir {}, deleted:{} ", stagingDir, isStgDeleted); |
| } |
| } |
| |
| public synchronized TezClient getClient(String appIdStr) throws IOException, TezException { |
| return getClient(appIdfromString(appIdStr)); |
| } |
| |
| /** |
| * Alternative to start() that explicitly sets sessionAppId and doesn't start a new AM. |
| * The caller of getClient is responsible for initializing the new TezClient with a |
| * Configuration compatible with the existing AM. It is expected the caller has cached the |
| * original Configuration (e.g. in Zookeeper). |
| * |
| * In contrast to "start", no resources are localized. It is the responsibility of the caller to |
| * ensure that existing localized resources and staging dirs are still valid. |
| * |
| * @param appId |
| * @return 'this' just as a convenience for fluent style chaining |
| */ |
| public synchronized TezClient getClient(ApplicationId appId) throws TezException, IOException { |
| sessionAppId = appId; |
| startFrameworkClient(); |
| setupJavaOptsChecker(); |
| |
| if (!isSession) { |
| String msg = "Must be in session mode to bind TezClient to existing AM"; |
| LOG.error(msg); |
| throw new IllegalStateException(msg); |
| } |
| |
| LOG.info("Session mode. Reconnecting to session: " + sessionAppId.toString()); |
| |
| clientTimeout = amConfig.getTezConfiguration().getInt( |
| TezConfiguration.TEZ_SESSION_CLIENT_TIMEOUT_SECS, |
| TezConfiguration.TEZ_SESSION_CLIENT_TIMEOUT_SECS_DEFAULT); |
| |
| try { |
| setupApplicationContext(); |
| ApplicationReport appReport = frameworkClient.getApplicationReport(sessionAppId); |
| LOG.info("The url to track the Tez Session: " + appReport.getTrackingUrl()); |
| sessionStarted.set(true); |
| } catch (YarnException e) { |
| cleanStagingDir(); |
| throw new TezException(e); |
| } catch (IOException e) { |
| cleanStagingDir(); |
| throw new TezException(e); |
| } |
| |
| startClientHeartbeat(); |
| this.stagingFs = FileSystem.get(amConfig.getTezConfiguration()); |
| return this; |
| } |
| |
| private void startFrameworkClient() { |
| frameworkClient = createFrameworkClient(); |
| frameworkClient.init(amConfig.getTezConfiguration()); |
| frameworkClient.start(); |
| } |
| |
| private ApplicationSubmissionContext setupApplicationContext() throws IOException, YarnException { |
| TezClientUtils.processTezLocalCredentialsFile(sessionCredentials, |
| amConfig.getTezConfiguration()); |
| |
| Map<String, LocalResource> tezJarResources = getTezJarResources(sessionCredentials); |
| // Add session token for shuffle |
| TezClientUtils.createSessionToken(sessionAppId.toString(), |
| jobTokenSecretManager, sessionCredentials); |
| |
| ApplicationSubmissionContext appContext = |
| TezClientUtils.createApplicationSubmissionContext( |
| sessionAppId, |
| null, clientName, amConfig, |
| tezJarResources, sessionCredentials, usingTezArchiveDeploy, apiVersionInfo, |
| servicePluginsDescriptor, javaOptsChecker); |
| |
| // Set Tez Sessions to not retry on AM crashes if recovery is disabled |
| if (!amConfig.getTezConfiguration().getBoolean( |
| TezConfiguration.DAG_RECOVERY_ENABLED, |
| TezConfiguration.DAG_RECOVERY_ENABLED_DEFAULT)) { |
| appContext.setMaxAppAttempts(1); |
| } |
| return appContext; |
| } |
| |
| private void setupJavaOptsChecker() { |
| if (this.amConfig.getTezConfiguration().getBoolean( |
| TezConfiguration.TEZ_CLIENT_JAVA_OPTS_CHECKER_ENABLED, |
| TezConfiguration.TEZ_CLIENT_JAVA_OPTS_CHECKER_ENABLED_DEFAULT)) { |
| String javaOptsCheckerClassName = this.amConfig.getTezConfiguration().get( |
| TezConfiguration.TEZ_CLIENT_JAVA_OPTS_CHECKER_CLASS, ""); |
| if (!javaOptsCheckerClassName.isEmpty()) { |
| try { |
| javaOptsChecker = ReflectionUtils.createClazzInstance(javaOptsCheckerClassName); |
| } catch (Exception e) { |
| LOG.warn("Failed to initialize configured Java Opts Checker" |
| + " (" + TezConfiguration.TEZ_CLIENT_JAVA_OPTS_CHECKER_CLASS |
| + ") , checkerClass=" + javaOptsCheckerClassName |
| + ". Disabling checker.", e); |
| javaOptsChecker = null; |
| } |
| } else { |
| javaOptsChecker = new JavaOptsChecker(); |
| } |
| |
| } |
| } |
| |
| private void startClientHeartbeat() { |
| long amClientKeepAliveTimeoutIntervalMillis = |
| TezCommonUtils.getAMClientHeartBeatTimeoutMillis(amConfig.getTezConfiguration()); |
| // Poll at minimum of 1 second interval |
| long pollPeriod = TezCommonUtils. |
| getAMClientHeartBeatPollIntervalMillis(amConfig.getTezConfiguration(), |
| amClientKeepAliveTimeoutIntervalMillis, 10); |
| |
| boolean isLocal = amConfig.getTezConfiguration().getBoolean( |
| TezConfiguration.TEZ_LOCAL_MODE, TezConfiguration.TEZ_LOCAL_MODE_DEFAULT); |
| if (!isLocal && amClientKeepAliveTimeoutIntervalMillis > 0) { |
| amKeepAliveService = Executors.newSingleThreadScheduledExecutor( |
| new ThreadFactoryBuilder() |
| .setDaemon(true).setNameFormat("AMKeepAliveThread #%d").build()); |
| amKeepAliveService.scheduleWithFixedDelay(new Runnable() { |
| |
| private DAGClientAMProtocolBlockingPB proxy; |
| |
| @Override |
| public void run() { |
| proxy = sendAMHeartbeat(proxy); |
| } |
| }, pollPeriod, pollPeriod, TimeUnit.MILLISECONDS); |
| } |
| } |
| |
| public DAGClientAMProtocolBlockingPB sendAMHeartbeat(DAGClientAMProtocolBlockingPB proxy) { |
| if (sessionStopped.get()) { |
| // Ignore sending heartbeat as session being stopped |
| return null; |
| } |
| try { |
| if (proxy == null) { |
| try { |
| proxy = frameworkClient.waitForProxy(clientTimeout, amConfig.getTezConfiguration(), |
| sessionAppId, getUgi()); |
| } catch (InterruptedException e) { |
| LOG.debug("Interrupted while trying to create a connection to the AM", e); |
| } catch (SessionNotRunning e) { |
| LOG.error("Cannot create a connection to the AM, stopping heartbeat to AM", e); |
| cancelAMKeepAlive(false); |
| } |
| } |
| if (proxy != null) { |
| LOG.debug("Sending heartbeat to AM"); |
| proxy.getAMStatus(null, GetAMStatusRequestProto.newBuilder().build()); |
| } |
| return proxy; |
| } catch (Exception e) { |
| LOG.info("Exception when sending heartbeat to AM for app {}: {}", sessionAppId, |
| e.getMessage()); |
| LOG.debug("Error when sending heartbeat ping to AM. Resetting AM proxy for app: {}" |
| + " due to exception :", sessionAppId, e); |
| return null; |
| } |
| } |
| |
| /** |
| * Submit a DAG. <br>In non-session mode, it submits a new App Master to the |
| * cluster.<br>In session mode, it submits the DAG to the session App Master. It |
| * blocks until either the DAG is submitted to the session or configured |
| * timeout period expires. Cleans up session if the submission timed out. |
| * |
| * @param dag |
| * DAG to be submitted to Session |
| * @return DAGClient to monitor the DAG |
| * @throws TezException |
| * @throws IOException |
| * @throws DAGSubmissionTimedOut |
| * if submission timed out |
| */ |
| public synchronized DAGClient submitDAG(DAG dag) throws TezException, IOException { |
| DAGClient result = isSession ? submitDAGSession(dag) : submitDAGApplication(dag); |
| if (result != null) { |
| closePrewarmDagClient(); // Assume the current DAG replaced the prewarm one; no need to kill. |
| } |
| return result; |
| } |
| |
| private void killAndClosePrewarmDagClient(long waitTimeMs) { |
| if (prewarmDagClient == null) { |
| return; |
| } |
| try { |
| prewarmDagClient.tryKillDAG(); |
| if (waitTimeMs > 0) { |
| LOG.info("Waiting for prewarm DAG to shut down"); |
| prewarmDagClient.waitForCompletion(waitTimeMs); |
| } |
| } |
| catch (Exception ex) { |
| LOG.warn("Failed to shut down the prewarm DAG " + prewarmDagClient, ex); |
| } |
| closePrewarmDagClient(); |
| } |
| |
| private void closePrewarmDagClient() { |
| if (prewarmDagClient == null) { |
| return; |
| } |
| try { |
| prewarmDagClient.close(); |
| } catch (Exception e) { |
| LOG.warn("Failed to close prewarm DagClient " + prewarmDagClient, e); |
| } |
| prewarmDagClient = null; |
| } |
| |
| private DAGClient submitDAGSession(DAG dag) throws TezException, IOException { |
| Preconditions.checkState(isSession == true, |
| "submitDAG with additional resources applies to only session mode. " + |
| "In non-session mode please specify all resources in the initial configuration"); |
| |
| verifySessionStateForSubmission(); |
| |
| String callerContextStr = ""; |
| if (dag.getCallerContext() != null) { |
| callerContextStr = ", callerContext=" + dag.getCallerContext().contextAsSimpleString(); |
| } |
| LOG.info("Submitting dag to TezSession" |
| + ", sessionName=" + clientName |
| + ", applicationId=" + sessionAppId |
| + ", dagName=" + dag.getName() |
| + callerContextStr); |
| |
| if (!additionalLocalResources.isEmpty()) { |
| for (LocalResource lr : additionalLocalResources.values()) { |
| Preconditions.checkArgument(lr.getType() == LocalResourceType.FILE, "LocalResourceType: " |
| + lr.getType() + " is not supported, only " + LocalResourceType.FILE + " is supported"); |
| } |
| } |
| |
| Map<String, LocalResource> tezJarResources = getTezJarResources(sessionCredentials); |
| DAGPlan dagPlan = TezClientUtils.prepareAndCreateDAGPlan(dag, amConfig, tezJarResources, |
| usingTezArchiveDeploy, sessionCredentials, servicePluginsDescriptor, javaOptsChecker); |
| |
| SubmitDAGRequestProto.Builder requestBuilder = SubmitDAGRequestProto.newBuilder(); |
| requestBuilder.setDAGPlan(dagPlan); |
| if (!additionalLocalResources.isEmpty()) { |
| requestBuilder.setAdditionalAmResources(DagTypeConverters |
| .convertFromLocalResources(additionalLocalResources)); |
| } |
| |
| additionalLocalResources.clear(); |
| |
| // if request size exceeds maxSubmitDAGRequestSizeThroughIPC, we serialize them to HDFS |
| SubmitDAGRequestProto request = requestBuilder.build(); |
| if (request.getSerializedSize() > maxSubmitDAGRequestSizeThroughIPC) { |
| Path dagPlanPath = new Path(TezCommonUtils.getTezSystemStagingPath(amConfig.getTezConfiguration(), |
| sessionAppId.toString()), TezConstants.TEZ_PB_PLAN_BINARY_NAME + |
| serializedSubmitDAGPlanRequestCounter.incrementAndGet()); |
| |
| FileSystem fs = dagPlanPath.getFileSystem(stagingFs.getConf()); |
| try (FSDataOutputStream fsDataOutputStream = fs.create(dagPlanPath, false)) { |
| LOG.info("Send dag plan using YARN local resources since it's too large" |
| + ", dag plan size=" + request.getSerializedSize() |
| + ", max dag plan size through IPC=" + maxSubmitDAGRequestSizeThroughIPC |
| + ", max IPC message size= " + amConfig.getTezConfiguration().getInt( |
| CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT)); |
| request.writeTo(fsDataOutputStream); |
| request = requestBuilder.clear().setSerializedRequestPath(fs.resolvePath(dagPlanPath).toString()).build(); |
| } |
| } |
| |
| return frameworkClient.submitDag(dag, request, clientName, sessionAppId, clientTimeout, |
| getUgi(), amConfig.getTezConfiguration()); |
| } |
| |
| private UserGroupInformation getUgi() throws IOException { |
| String userName = UserGroupInformation.getCurrentUser().getUserName(); |
| return ugiMap.computeIfAbsent(userName, |
| v -> UserGroupInformation.createRemoteUser(userName)); |
| } |
| |
| @VisibleForTesting |
| protected long getPrewarmWaitTimeMs() { |
| return PREWARM_WAIT_MS; |
| } |
| |
| /** |
| * Stop the client. This terminates the connection to the YARN cluster. |
| * In session mode, this shuts down the session DAG App Master |
| * @throws TezException |
| * @throws IOException |
| */ |
| public synchronized void stop() throws TezException, IOException { |
| killAndClosePrewarmDagClient(getPrewarmWaitTimeMs()); |
| try { |
| if (amKeepAliveService != null) { |
| amKeepAliveService.shutdownNow(); |
| } |
| if (sessionStarted.get()) { |
| LOG.info("Shutting down Tez Session" |
| + ", sessionName=" + clientName |
| + ", applicationId=" + sessionAppId); |
| sessionStopped.set(true); |
| boolean sessionShutdownSuccessful = false; |
| try { |
| sessionShutdownSuccessful = frameworkClient |
| .shutdownSession(amConfig.getTezConfiguration(), sessionAppId, getUgi()); |
| boolean asynchronousStop = amConfig.getTezConfiguration().getBoolean( |
| TezConfiguration.TEZ_CLIENT_ASYNCHRONOUS_STOP, |
| TezConfiguration.TEZ_CLIENT_ASYNCHRONOUS_STOP_DEFAULT); |
| if (!asynchronousStop && sessionShutdownSuccessful) { |
| LOG.info("Waiting until application is in a final state"); |
| long currentTimeMillis = System.currentTimeMillis(); |
| long timeKillIssued = currentTimeMillis; |
| long killTimeOut = amConfig.getTezConfiguration().getLong( |
| TezConfiguration.TEZ_CLIENT_HARD_KILL_TIMEOUT_MS, |
| TezConfiguration.TEZ_CLIENT_HARD_KILL_TIMEOUT_MS_DEFAULT); |
| ApplicationReport appReport = frameworkClient |
| .getApplicationReport(sessionAppId); |
| while ((currentTimeMillis < timeKillIssued + killTimeOut) |
| && !isJobInTerminalState(appReport.getYarnApplicationState())) { |
| try { |
| Thread.sleep(1000L); |
| } catch (InterruptedException ie) { |
| /** interrupted, just break */ |
| break; |
| } |
| currentTimeMillis = System.currentTimeMillis(); |
| appReport = frameworkClient.getApplicationReport(sessionAppId); |
| } |
| |
| if (!isJobInTerminalState(appReport.getYarnApplicationState())) { |
| frameworkClient.killApplication(sessionAppId); |
| } |
| } |
| } catch (TezException e) { |
| LOG.info("Failed to shutdown Tez Session via proxy", e); |
| } catch (ServiceException e) { |
| LOG.info("Failed to shutdown Tez Session via proxy", e); |
| } catch (ApplicationNotFoundException e) { |
| LOG.info("Failed to kill nonexistent application " + sessionAppId, e); |
| } catch (YarnException e) { |
| throw new TezException(e); |
| } |
| if (!sessionShutdownSuccessful) { |
| LOG.info("Could not connect to AM, killing session via YARN" |
| + ", sessionName=" + clientName |
| + ", applicationId=" + sessionAppId); |
| try { |
| frameworkClient.killApplication(sessionAppId); |
| } catch (ApplicationNotFoundException e) { |
| LOG.info("Failed to kill nonexistent application " + sessionAppId, e); |
| } catch (YarnException e) { |
| throw new TezException(e); |
| } |
| } |
| } |
| } finally { |
| if (frameworkClient != null) { |
| frameworkClient.close(); |
| } |
| } |
| } |
| private boolean isJobInTerminalState(YarnApplicationState yarnApplicationState) { |
| return (yarnApplicationState == YarnApplicationState.FINISHED |
| || yarnApplicationState == YarnApplicationState.FAILED |
| || yarnApplicationState == YarnApplicationState.KILLED); |
| } |
| |
| /** |
| * Get the name of the client |
| * @return name |
| */ |
| public String getClientName() { |
| return clientName; |
| } |
| |
| @Private |
| @VisibleForTesting |
| public synchronized ApplicationId getAppMasterApplicationId() { |
| if (isSession) { |
| return sessionAppId; |
| } else { |
| return lastSubmittedAppId; |
| } |
| } |
| |
| /** |
| * Get the status of the App Master executing the DAG |
| * In non-session mode it returns the status of the last submitted DAG App Master |
| * In session mode, it returns the status of the App Master hosting the session |
| * |
| * @return State of the session |
| * @throws TezException |
| * @throws IOException |
| */ |
| public synchronized TezAppMasterStatus getAppMasterStatus() throws TezException, IOException { |
| // Supporting per-DAG app master case since user may choose to run the same |
| // code in that mode and the code should continue to work. Its easy to provide |
| // the correct view for per-DAG app master too. |
| ApplicationId appId = null; |
| if (isSession) { |
| appId = sessionAppId; |
| } else { |
| appId = lastSubmittedAppId; |
| } |
| Preconditions.checkState(appId != null, "Cannot get status without starting an application"); |
| try { |
| ApplicationReport appReport = frameworkClient.getApplicationReport( |
| appId); |
| switch (appReport.getYarnApplicationState()) { |
| case NEW: |
| case NEW_SAVING: |
| case ACCEPTED: |
| case SUBMITTED: |
| return TezAppMasterStatus.INITIALIZING; |
| case FAILED: |
| case KILLED: |
| diagnostics = appReport.getDiagnostics(); |
| LOG.info("App did not succeed. Diagnostics: " |
| + (appReport.getDiagnostics() != null ? appReport.getDiagnostics() |
| : NO_CLUSTER_DIAGNOSTICS_MSG)); |
| return TezAppMasterStatus.SHUTDOWN; |
| case FINISHED: |
| return TezAppMasterStatus.SHUTDOWN; |
| case RUNNING: |
| try { |
| return frameworkClient.getAMStatus(amConfig.getTezConfiguration(), appId, getUgi()); |
| } catch (TezException e) { |
| LOG.info("Failed to retrieve AM Status via proxy", e); |
| } catch (ServiceException e) { |
| LOG.info("Failed to retrieve AM Status via proxy", e); |
| } |
| } |
| } catch (ApplicationNotFoundException e) { |
| return TezAppMasterStatus.SHUTDOWN; |
| } catch (YarnException e) { |
| throw new TezException(e); |
| } |
| return TezAppMasterStatus.INITIALIZING; |
| } |
| |
| /** |
| * API to help pre-allocate containers in session mode. In non-session mode |
| * this is ignored. The pre-allocated containers may be re-used by subsequent |
| * job DAGs to improve performance. |
| * The preWarm vertex should be configured and setup exactly |
| * like the other vertices in the job DAGs so that the pre-allocated containers |
| * may be re-used by the subsequent DAGs to improve performance. |
| * The processor for the preWarmVertex may be used to pre-warm the containers |
| * by pre-loading classes etc. It should be short-running so that pre-warming |
| * does not block real execution. Users can specify their custom processors or |
| * use the PreWarmProcessor from the runtime library. |
| * The parallelism of the preWarmVertex will determine the number of preWarmed |
| * containers. |
| * Pre-warming is best efforts and among other factors is limited by the free |
| * resources on the cluster. |
| * @param preWarmVertex |
| * @throws TezException |
| * @throws IOException |
| */ |
| @Unstable |
| public synchronized void preWarm(PreWarmVertex preWarmVertex) throws TezException, IOException { |
| preWarm(preWarmVertex, 0, TimeUnit.MILLISECONDS); |
| } |
| |
| /** |
| * API to help pre-allocate containers in session mode. In non-session mode |
| * this is ignored. The pre-allocated containers may be re-used by subsequent |
| * job DAGs to improve performance. |
| * The preWarm vertex should be configured and setup exactly |
| * like the other vertices in the job DAGs so that the pre-allocated |
| * containers may be re-used by the subsequent DAGs to improve performance. |
| * The processor for the preWarmVertex may be used to pre-warm the containers |
| * by pre-loading classes etc. It should be short-running so that pre-warming |
| * does not block real execution. Users can specify their custom processors or |
| * use the PreWarmProcessor from the runtime library. |
| * The parallelism of the preWarmVertex will determine the number of preWarmed |
| * containers. |
| * Pre-warming is best efforts and among other factors is limited by the free |
| * resources on the cluster. Based on the specified timeout value it returns |
| * false if the status is not READY after the wait period. |
| * @param preWarmVertex |
| * @param timeout |
| * @param unit |
| * @throws TezException |
| * @throws IOException |
| */ |
| @Unstable |
| public synchronized void preWarm(PreWarmVertex preWarmVertex, |
| long timeout, TimeUnit unit) |
| throws TezException, IOException { |
| if (!isSession) { |
| // do nothing for non session mode. This is there to let the code |
| // work correctly in both modes |
| LOG.warn("preWarm is not supported in non-session mode," + |
| "please use session-mode of TezClient"); |
| return; |
| } |
| |
| verifySessionStateForSubmission(); |
| |
| DAG dag = org.apache.tez.dag.api.DAG.create(TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX + "_" |
| + preWarmDAGCounter++); |
| dag.addVertex(preWarmVertex); |
| |
| boolean isReady; |
| try { |
| isReady = waitTillReady(timeout, unit); |
| } catch (InterruptedException e) { |
| throw new IOException("Interrupted while waiting for AM to become " + |
| "available", e); |
| } |
| if(isReady) { |
| prewarmDagClient = submitDAG(dag); |
| } else { |
| throw new SessionNotReady("Tez AM not ready, could not submit DAG"); |
| } |
| } |
| |
| |
| /** |
| * Wait till the DAG is ready to be submitted. |
| * In non-session mode this is a no-op since the application can be immediately |
| * submitted. |
| * In session mode, this waits for the session host to be ready to accept a DAG |
| * @throws IOException |
| * @throws TezException |
| * @throws InterruptedException |
| */ |
| @Evolving |
| public synchronized void waitTillReady() throws IOException, TezException, InterruptedException { |
| waitTillReady(0, TimeUnit.MILLISECONDS); |
| } |
| |
| /** |
| * Wait till the DAG is ready to be submitted. |
| * In non-session mode this is a no-op since the application can be |
| * immediately submitted. |
| * In session mode, this waits for the session host to be ready to accept |
| * a DAG and returns false if not ready after a configured time wait period. |
| * @param timeout |
| * @param unit |
| * @return true if READY or is not in session mode, false otherwise. |
| * @throws IOException |
| * @throws TezException |
| * @throws InterruptedException |
| */ |
| @Evolving |
| public synchronized boolean waitTillReady(long timeout, TimeUnit unit) |
| throws IOException, TezException, InterruptedException { |
| timeout = unit.toMillis(timeout); |
| if (!isSession) { |
| // nothing to wait for in non-session mode |
| return true; |
| } |
| |
| verifySessionStateForSubmission(); |
| long startTime = Time.monotonicNow(); |
| long timeLimit = startTime + timeout; |
| while (true) { |
| TezAppMasterStatus status = getAppMasterStatus(); |
| if (status.equals(TezAppMasterStatus.SHUTDOWN)) { |
| throw new SessionNotRunning("TezSession has already shutdown. " |
| + ((diagnostics != null) ? diagnostics : NO_CLUSTER_DIAGNOSTICS_MSG)); |
| } |
| if (status.equals(TezAppMasterStatus.READY)) { |
| return true; |
| } |
| if (timeout == 0) { |
| Thread.sleep(SLEEP_FOR_READY); |
| continue; |
| } |
| long now = Time.monotonicNow(); |
| if (timeLimit > now) { |
| long sleepTime = Math.min(SLEEP_FOR_READY, timeLimit - now); |
| Thread.sleep(sleepTime); |
| } else { |
| return false; |
| } |
| } |
| } |
| |
| private void waitNonSessionTillReady() throws IOException, TezException { |
| Preconditions.checkArgument(!isSession, "It is supposed to be only called in non-session mode"); |
| while (true) { |
| TezAppMasterStatus status = getAppMasterStatus(); |
| // DAGClient will handle the AM SHUTDOWN case |
| if (status.equals(TezAppMasterStatus.RUNNING) |
| || status.equals(TezAppMasterStatus.SHUTDOWN)) { |
| return; |
| } |
| try { |
| Thread.sleep(SLEEP_FOR_READY); |
| } catch (InterruptedException e) { |
| throw new TezException("TezClient is interrupted"); |
| } |
| } |
| } |
| @VisibleForTesting |
| // for testing |
| @Private |
| protected FrameworkClient createFrameworkClient() { |
| return FrameworkClient.createFrameworkClient(amConfig.getTezConfiguration()); |
| } |
| |
| private void verifySessionStateForSubmission() throws SessionNotRunning { |
| Preconditions.checkState(isSession, "Invalid without session mode"); |
| if (!sessionStarted.get()) { |
| throw new SessionNotRunning("Session not started"); |
| } else if (sessionStopped.get()) { |
| throw new SessionNotRunning("Session stopped by user"); |
| } |
| } |
| |
| private DAGClient submitDAGApplication(DAG dag) |
| throws TezException, IOException { |
| ApplicationId appId = createApplication(); |
| return submitDAGApplication(appId, dag); |
| } |
| |
| @Private // To be used only by YarnRunner |
| DAGClient submitDAGApplication(ApplicationId appId, DAG dag) |
| throws TezException, IOException { |
| LOG.info("Submitting DAG application with id: " + appId); |
| try { |
| // Use the AMCredentials object in client mode, since this won't be re-used. |
| // Ensures we don't fetch credentially unnecessarily if the user has already provided them. |
| Credentials credentials = amConfig.getCredentials(); |
| if (credentials == null) { |
| credentials = new Credentials(); |
| } |
| TezClientUtils.processTezLocalCredentialsFile(credentials, |
| amConfig.getTezConfiguration()); |
| |
| // Add session token for shuffle |
| TezClientUtils.createSessionToken(appId.toString(), |
| jobTokenSecretManager, credentials); |
| |
| // Add credentials for tez-local resources. |
| Map<String, LocalResource> tezJarResources = getTezJarResources(credentials); |
| ApplicationSubmissionContext appContext = TezClientUtils |
| .createApplicationSubmissionContext( |
| appId, dag, dag.getName(), amConfig, tezJarResources, credentials, |
| usingTezArchiveDeploy, apiVersionInfo, servicePluginsDescriptor, javaOptsChecker); |
| String callerContextStr = ""; |
| if (dag.getCallerContext() != null) { |
| callerContextStr = ", callerContext=" + dag.getCallerContext().contextAsSimpleString(); |
| } |
| LOG.info("Submitting DAG to YARN" |
| + ", applicationId=" + appId |
| + ", dagName=" + dag.getName() |
| + callerContextStr); |
| TezCommonUtils.logCredentials(LOG, credentials, "appContext"); |
| frameworkClient.submitApplication(appContext); |
| ApplicationReport appReport = frameworkClient.getApplicationReport(appId); |
| LOG.info("The url to track the Tez AM: " + appReport.getTrackingUrl()); |
| lastSubmittedAppId = appId; |
| } catch (YarnException e) { |
| throw new TezException(e); |
| } |
| // wait for dag in non-session mode to start running, so that we can start to getDAGStatus |
| waitNonSessionTillReady(); |
| return getDAGClient(appId, amConfig.getTezConfiguration(), frameworkClient, getUgi()); |
| } |
| |
| private ApplicationId createApplication() throws TezException, IOException { |
| try { |
| return frameworkClient.createApplication(). |
| getNewApplicationResponse().getApplicationId(); |
| } catch (YarnException e) { |
| throw new TezException(e); |
| } |
| } |
| |
| private synchronized Map<String, LocalResource> getTezJarResources(Credentials credentials) |
| throws IOException { |
| if (cachedTezJarResources == null) { |
| cachedTezJarResources = new HashMap<String, LocalResource>(); |
| usingTezArchiveDeploy = TezClientUtils.setupTezJarsLocalResources( |
| amConfig.getTezConfiguration(), credentials, cachedTezJarResources); |
| } |
| return cachedTezJarResources; |
| } |
| |
| @Private |
| static DAGClient getDAGClient(ApplicationId appId, TezConfiguration tezConf, |
| FrameworkClient frameworkClient, UserGroupInformation ugi) throws IOException, TezException { |
| return frameworkClient.getDAGClient(appId, getDefaultTezDAGID(appId), tezConf, ugi); |
| } |
| |
| // DO NOT CHANGE THIS. This code is replicated from TezDAGID.java |
| private static final char SEPARATOR = '_'; |
| public static final String DAG = "dag"; |
| static final ThreadLocal<NumberFormat> tezAppIdFormat = new ThreadLocal<NumberFormat>() { |
| @Override |
| public NumberFormat initialValue() { |
| NumberFormat fmt = NumberFormat.getInstance(); |
| fmt.setGroupingUsed(false); |
| fmt.setMinimumIntegerDigits(4); |
| return fmt; |
| } |
| }; |
| |
| static final ThreadLocal<NumberFormat> tezDagIdFormat = new ThreadLocal<NumberFormat>() { |
| @Override |
| public NumberFormat initialValue() { |
| NumberFormat fmt = NumberFormat.getInstance(); |
| fmt.setGroupingUsed(false); |
| fmt.setMinimumIntegerDigits(1); |
| return fmt; |
| } |
| }; |
| |
| // Used only for MapReduce compatibility code |
| private static String getDefaultTezDAGID(ApplicationId applicationId) { |
| return (new StringBuilder(DAG)).append(SEPARATOR). |
| append(applicationId.getClusterTimestamp()). |
| append(SEPARATOR). |
| append(tezAppIdFormat.get().format(applicationId.getId())). |
| append(SEPARATOR). |
| append(tezDagIdFormat.get().format(1)).toString(); |
| } |
| |
| @VisibleForTesting |
| @Private |
| public synchronized void cancelAMKeepAlive(boolean shutdownNow) { |
| if (amKeepAliveService != null) { |
| if (shutdownNow) { |
| amKeepAliveService.shutdownNow(); |
| } else { |
| amKeepAliveService.shutdown(); |
| } |
| } |
| } |
| |
| @VisibleForTesting |
| protected synchronized ScheduledExecutorService getAMKeepAliveService() { |
| return amKeepAliveService; |
| } |
| |
| /** |
| * A builder for setting up an instance of {@link org.apache.tez.client.TezClient} |
| */ |
| @Public |
| public static class TezClientBuilder { |
| final String name; |
| final TezConfiguration tezConf; |
| boolean isSession; |
| private Map<String, LocalResource> localResourceMap; |
| private Credentials credentials; |
| ServicePluginsDescriptor servicePluginsDescriptor; |
| |
| /** |
| * Create an instance of a TezClientBuilder |
| * |
| * @param name |
| * Name of the client. Used for logging etc. This will also be used |
| * as app master name is session mode |
| * @param tezConf |
| * Configuration for the framework |
| */ |
| private TezClientBuilder(String name, TezConfiguration tezConf) { |
| this.name = name; |
| this.tezConf = tezConf; |
| isSession = tezConf.getBoolean( |
| TezConfiguration.TEZ_AM_SESSION_MODE, TezConfiguration.TEZ_AM_SESSION_MODE_DEFAULT); |
| } |
| |
| /** |
| * Specify whether this client is a session or not |
| * @param isSession whether the client is a session |
| * @return the current builder |
| */ |
| public TezClientBuilder setIsSession(boolean isSession) { |
| this.isSession = isSession; |
| return this; |
| } |
| |
| /** |
| * Set local resources to be used by the AppMaster |
| * |
| * @param localResources local files for the App Master |
| * @return the files to be added to the AM |
| */ |
| public TezClientBuilder setLocalResources(Map<String, LocalResource> localResources) { |
| this.localResourceMap = localResources; |
| return this; |
| } |
| |
| /** |
| * Setup security credentials |
| * |
| * @param credentials |
| * Set security credentials to be used inside the app master, if |
| * needed. Tez App Master needs credentials to access the staging |
| * directory and for most HDFS cases these are automatically obtained |
| * by Tez client. If the staging directory is on a file system for |
| * which credentials cannot be obtained or for any credentials needed |
| * by user code running inside the App Master, credentials must be |
| * supplied by the user. These will be used by the App Master for the |
| * next DAG. <br> |
| * In session mode, credentials, if needed, must be set before |
| * calling start() |
| * @return the current builder |
| */ |
| public TezClientBuilder setCredentials(Credentials credentials) { |
| this.credentials = credentials; |
| return this; |
| } |
| |
| /** |
| * Specify the service plugins that will be running in the AM |
| * @param servicePluginsDescriptor the service plugin descriptor with details about the plugins running in the AM |
| * @return the current builder |
| */ |
| public TezClientBuilder setServicePluginDescriptor(ServicePluginsDescriptor servicePluginsDescriptor) { |
| this.servicePluginsDescriptor = servicePluginsDescriptor; |
| return this; |
| } |
| |
| /** |
| * Build the actual instance of the {@link TezClient} |
| * @return an instance of {@link TezClient} |
| */ |
| public TezClient build() { |
| return new TezClient(name, tezConf, isSession, localResourceMap, credentials, |
| servicePluginsDescriptor); |
| } |
| } |
| |
| //Copied this helper method from |
| //org.apache.hadoop.yarn.api.records.ApplicationId in Hadoop 2.8+ |
| //to simplify implementation on 2.7.x |
| @Public |
| @Unstable |
| public static ApplicationId appIdfromString(String appIdStr) { |
| if (!appIdStr.startsWith(APPLICATION_ID_PREFIX)) { |
| throw new IllegalArgumentException("Invalid ApplicationId prefix: " |
| + appIdStr + ". The valid ApplicationId should start with prefix " |
| + appIdStrPrefix); |
| } |
| try { |
| int pos1 = APPLICATION_ID_PREFIX.length() - 1; |
| int pos2 = appIdStr.indexOf('_', pos1 + 1); |
| if (pos2 < 0) { |
| throw new IllegalArgumentException("Invalid ApplicationId: " |
| + appIdStr); |
| } |
| long rmId = Long.parseLong(appIdStr.substring(pos1 + 1, pos2)); |
| int appId = Integer.parseInt(appIdStr.substring(pos2 + 1)); |
| ApplicationId applicationId = ApplicationId.newInstance(rmId, appId); |
| return applicationId; |
| } catch (NumberFormatException n) { |
| throw new IllegalArgumentException("Invalid ApplicationId: " |
| + appIdStr, n); |
| } |
| } |
| } |
