You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "smallzhongfeng (via GitHub)" <gi...@apache.org> on 2023/05/11 07:56:36 UTC
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #871: [#870] feat(tez): Tez Shuffle Read supporting RemoteShuffle related app master and reader
smallzhongfeng commented on code in PR #871:
URL: https://github.com/apache/incubator-uniffle/pull/871#discussion_r1190769721
##########
client-tez/src/main/java/org/apache/tez/dag/app/RssDAGAppMaster.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.dag.app;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.Options;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.mapreduce.JobSubmissionFiles;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ShutdownHookManager;
+import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.SystemClock;
+import org.apache.tez.common.RssTezConfig;
+import org.apache.tez.common.RssTezUtils;
+import org.apache.tez.common.TezClassLoader;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtilsInternal;
+import org.apache.tez.common.VersionInfo;
+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.records.DAGProtos;
+import org.apache.tez.dag.api.records.DAGProtos.AMPluginDescriptorProto;
+import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+
+import static org.apache.tez.common.TezCommonUtils.TEZ_SYSTEM_SUB_DIR;
+
+public class RssDAGAppMaster extends DAGAppMaster {
+ private static final Logger LOG = LoggerFactory.getLogger(RssDAGAppMaster.class);
+ private ShuffleWriteClient shuffleWriteClient;
+ private TezRemoteShuffleManager tezRemoteShuffleManager;
+ private final String rssConfFileLocalResourceName = "rss_conf.xml";
+ private DAGProtos.PlanLocalResource rssConfFileLocalResource;
+ final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ return t;
+ }
+ }
+ );
+
+ public RssDAGAppMaster(ApplicationAttemptId applicationAttemptId,
+ ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
Review Comment:
nit: pay attention to indentation, 4 characters.
##########
client-tez/src/main/java/org/apache/tez/dag/app/RssDAGAppMaster.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.dag.app;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.Options;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.mapreduce.JobSubmissionFiles;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ShutdownHookManager;
+import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.SystemClock;
+import org.apache.tez.common.RssTezConfig;
+import org.apache.tez.common.RssTezUtils;
+import org.apache.tez.common.TezClassLoader;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtilsInternal;
+import org.apache.tez.common.VersionInfo;
+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.records.DAGProtos;
+import org.apache.tez.dag.api.records.DAGProtos.AMPluginDescriptorProto;
+import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+
+import static org.apache.tez.common.TezCommonUtils.TEZ_SYSTEM_SUB_DIR;
+
+public class RssDAGAppMaster extends DAGAppMaster {
+ private static final Logger LOG = LoggerFactory.getLogger(RssDAGAppMaster.class);
+ private ShuffleWriteClient shuffleWriteClient;
+ private TezRemoteShuffleManager tezRemoteShuffleManager;
+ private final String rssConfFileLocalResourceName = "rss_conf.xml";
+ private DAGProtos.PlanLocalResource rssConfFileLocalResource;
+ final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ return t;
+ }
+ }
+ );
+
+ public RssDAGAppMaster(ApplicationAttemptId applicationAttemptId,
+ ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
+ Clock clock, long appSubmitTime, boolean isSession, String workingDirectory,
+ String [] localDirs, String[] logDirs, String clientVersion,
+ Credentials credentials, String jobUserName, AMPluginDescriptorProto pluginDescriptorProto) {
+ super(applicationAttemptId, containerId, nmHost, nmPort, nmHttpPort, clock, appSubmitTime, isSession,
+ workingDirectory, localDirs, logDirs, clientVersion, credentials, jobUserName, pluginDescriptorProto);
+ }
+
+ public ShuffleWriteClient getShuffleWriteClient() {
+ return shuffleWriteClient;
+ }
+
+ public void setShuffleWriteClient(ShuffleWriteClient shuffleWriteClient) {
+ this.shuffleWriteClient = shuffleWriteClient;
+ }
+
+ public TezRemoteShuffleManager getTezRemoteShuffleManager() {
+ return tezRemoteShuffleManager;
+ }
+
+ public void setTezRemoteShuffleManager(TezRemoteShuffleManager tezRemoteShuffleManager) {
+ this.tezRemoteShuffleManager = tezRemoteShuffleManager;
+ }
+
+ /**
+ * Init and Start Rss Client
+ * @param appMaster
+ * @param conf
+ * @param applicationAttemptId
+ * @throws Exception
+ */
+ public static void initAndStartRSSClient(final RssDAGAppMaster appMaster, Configuration conf,
+ ApplicationAttemptId applicationAttemptId) throws Exception {
Review Comment:
nit: pay attention to indentation, 4 characters.
##########
client-tez/src/main/java/org/apache/tez/runtime/library/common/readers/RssUnorderedKVReader.java:
##########
@@ -0,0 +1,234 @@
+/**
+ * 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.runtime.library.common.readers;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.serializer.Deserializer;
+import org.apache.hadoop.io.serializer.SerializationFactory;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.IOInterruptedException;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.common.ConfigUtils;
+import org.apache.tez.runtime.library.common.shuffle.FetchedInput;
+import org.apache.tez.runtime.library.common.shuffle.FetchedInput.Type;
+import org.apache.tez.runtime.library.common.shuffle.MemoryFetchedInput;
+import org.apache.tez.runtime.library.common.shuffle.impl.ShuffleManager;
+import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.InMemoryReader;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Unstable
+@Private
+public class RssUnorderedKVReader<K, V> extends KeyValueReader {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UnorderedKVReader.class);
+
+ private final ShuffleManager shuffleManager;
+ private final CompressionCodec codec;
+
+ private final Class<K> keyClass;
+ private final Class<V> valClass;
+ private final Deserializer<K> keyDeserializer;
+ private final Deserializer<V> valDeserializer;
+ private final DataInputBuffer keyIn;
+ private final DataInputBuffer valIn;
+
+ private final boolean ifileReadAhead;
+ private final int ifileReadAheadLength;
+ private final int ifileBufferSize;
+
+ private final TezCounter inputRecordCounter;
+ private final InputContext context;
+
+ private K key;
+ private V value;
+
+ private FetchedInput currentFetchedInput;
+ private IFile.Reader currentReader;
+
+ // TODO Remove this once per I/O counters are separated properly. Relying on
+ // the counter at the moment will generate aggregate numbers.
+ private int numRecordsRead = 0;
+ private final AtomicLong totalBytesRead = new AtomicLong(0);
+ private final AtomicLong totalFileBytes = new AtomicLong(0);
+
+
+ public RssUnorderedKVReader(ShuffleManager shuffleManager, Configuration conf,
+ CompressionCodec codec, boolean ifileReadAhead, int ifileReadAheadLength, int ifileBufferSize,
+ TezCounter inputRecordCounter, InputContext context)
+ throws IOException {
+ this.shuffleManager = shuffleManager;
+ this.context = context;
+ this.codec = codec;
+ this.ifileReadAhead = ifileReadAhead;
+ this.ifileReadAheadLength = ifileReadAheadLength;
+ this.ifileBufferSize = ifileBufferSize;
+ this.inputRecordCounter = inputRecordCounter;
+
+ this.keyClass = ConfigUtils.getIntermediateInputKeyClass(conf);
+ this.valClass = ConfigUtils.getIntermediateInputValueClass(conf);
+
+ this.keyIn = new DataInputBuffer();
+ this.valIn = new DataInputBuffer();
+
+ SerializationFactory serializationFactory = new SerializationFactory(conf);
+
+ this.keyDeserializer = serializationFactory.getDeserializer(keyClass);
+ this.keyDeserializer.open(keyIn);
+ this.valDeserializer = serializationFactory.getDeserializer(valClass);
+ this.valDeserializer.open(valIn);
+ }
+
+ // TODO NEWTEZ Maybe add an interface to check whether next will block.
Review Comment:
If we need add a `TODO`, add a new issue also plz.
##########
client-tez/src/main/java/org/apache/tez/dag/app/RssDAGAppMaster.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.dag.app;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.Options;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.mapreduce.JobSubmissionFiles;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ShutdownHookManager;
+import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.SystemClock;
+import org.apache.tez.common.RssTezConfig;
+import org.apache.tez.common.RssTezUtils;
+import org.apache.tez.common.TezClassLoader;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtilsInternal;
+import org.apache.tez.common.VersionInfo;
+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.records.DAGProtos;
+import org.apache.tez.dag.api.records.DAGProtos.AMPluginDescriptorProto;
+import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+
+import static org.apache.tez.common.TezCommonUtils.TEZ_SYSTEM_SUB_DIR;
+
+public class RssDAGAppMaster extends DAGAppMaster {
+ private static final Logger LOG = LoggerFactory.getLogger(RssDAGAppMaster.class);
+ private ShuffleWriteClient shuffleWriteClient;
+ private TezRemoteShuffleManager tezRemoteShuffleManager;
+ private final String rssConfFileLocalResourceName = "rss_conf.xml";
+ private DAGProtos.PlanLocalResource rssConfFileLocalResource;
+ final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ return t;
+ }
+ }
+ );
+
+ public RssDAGAppMaster(ApplicationAttemptId applicationAttemptId,
+ ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
+ Clock clock, long appSubmitTime, boolean isSession, String workingDirectory,
+ String [] localDirs, String[] logDirs, String clientVersion,
+ Credentials credentials, String jobUserName, AMPluginDescriptorProto pluginDescriptorProto) {
+ super(applicationAttemptId, containerId, nmHost, nmPort, nmHttpPort, clock, appSubmitTime, isSession,
+ workingDirectory, localDirs, logDirs, clientVersion, credentials, jobUserName, pluginDescriptorProto);
+ }
+
+ public ShuffleWriteClient getShuffleWriteClient() {
+ return shuffleWriteClient;
+ }
+
+ public void setShuffleWriteClient(ShuffleWriteClient shuffleWriteClient) {
+ this.shuffleWriteClient = shuffleWriteClient;
+ }
+
+ public TezRemoteShuffleManager getTezRemoteShuffleManager() {
+ return tezRemoteShuffleManager;
+ }
+
+ public void setTezRemoteShuffleManager(TezRemoteShuffleManager tezRemoteShuffleManager) {
+ this.tezRemoteShuffleManager = tezRemoteShuffleManager;
+ }
+
+ /**
+ * Init and Start Rss Client
+ * @param appMaster
+ * @param conf
+ * @param applicationAttemptId
+ * @throws Exception
+ */
+ public static void initAndStartRSSClient(final RssDAGAppMaster appMaster, Configuration conf,
+ ApplicationAttemptId applicationAttemptId) throws Exception {
+
+ ShuffleWriteClient client = RssTezUtils.createShuffleClient(conf);
+ appMaster.setShuffleWriteClient(client);
+
+ String coordinators = conf.get(RssTezConfig.RSS_COORDINATOR_QUORUM);
+ LOG.info("Registering coordinators {}", coordinators);
+ client.registerCoordinators(coordinators);
+
+ String strAppAttemptId = applicationAttemptId.toString();
+ long heartbeatInterval = conf.getLong(RssTezConfig.RSS_HEARTBEAT_INTERVAL,
+ RssTezConfig.RSS_HEARTBEAT_INTERVAL_DEFAULT_VALUE);
+ long heartbeatTimeout = conf.getLong(RssTezConfig.RSS_HEARTBEAT_TIMEOUT, heartbeatInterval / 2);
+ client.registerApplicationInfo(strAppAttemptId, heartbeatTimeout, "user");
+
+ appMaster.scheduledExecutorService.scheduleAtFixedRate(
+ () -> {
+ try {
+ client.sendAppHeartbeat(strAppAttemptId, heartbeatTimeout);
+ LOG.info("Finish send heartbeat to coordinator and servers");
+ } catch (Exception e) {
+ LOG.warn("Fail to send heartbeat to coordinator and servers", e);
+ }
+ },
+ heartbeatInterval / 2,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+
+ appMaster.setTezRemoteShuffleManager(new TezRemoteShuffleManager(strAppAttemptId, null, conf,
+ strAppAttemptId, client));
+ appMaster.getTezRemoteShuffleManager().initialize();
+ appMaster.getTezRemoteShuffleManager().start();
+
+ TezConfiguration extraConf = new TezConfiguration(false);
+ extraConf.clear();
+
+ String strAppId = applicationAttemptId.getApplicationId().toString();
+ extraConf.set(RssTezConfig.RSS_AM_SHUFFLE_MANAGER_ADDRESS,
+ appMaster.getTezRemoteShuffleManager().address.getHostName());
+ extraConf.setInt(RssTezConfig.RSS_AM_SHUFFLE_MANAGER_PORT,
+ appMaster.getTezRemoteShuffleManager().address.getPort());
+ writeExtraConf(appMaster, conf, extraConf, strAppId);
+
+ mayCloseTezSlowStart(conf);
+ }
+
+ @Override
+ public String submitDAGToAppMaster(DAGProtos.DAGPlan dagPlan,
+ Map<String, LocalResource> additionalResources) throws TezException {
+ List<DAGProtos.PlanLocalResource> planLocalResourceList = dagPlan.getLocalResourceList();
+
+ if (planLocalResourceList == null) {
+ LOG.warn("planLocalResourceList is null, add new list");
+ planLocalResourceList = new ArrayList<>();
+ } else {
+ planLocalResourceList = new ArrayList<>(planLocalResourceList);
+ }
+
+ try {
+ planLocalResourceList.add(rssConfFileLocalResource);
+ Field field = DAGProtos.DAGPlan.class.getDeclaredField("localResource_");
+ field.setAccessible(true);
+ field.set(dagPlan, planLocalResourceList);
+ field.setAccessible(false);
+ } catch (Exception e) {
+ LOG.error("submitDAGToAppMaster reflect error", e);
+ throw new TezException(e.getMessage());
+ }
+
+ if (LOG.isDebugEnabled()) {
+ for (DAGProtos.PlanLocalResource localResource : dagPlan.getLocalResourceList()) {
+ LOG.debug("localResource: {}", localResource.toString());
+ }
+ }
+
+ return super.submitDAGToAppMaster(dagPlan, additionalResources);
+ }
+
+ /**
+ * main method
+ * @param args
+ */
+ public static void main(String[] args) {
+ try {
+ // Install the tez class loader, which can be used add new resources
+ TezClassLoader.setupTezClassLoader();
+ Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
+ final String pid = System.getenv().get("JVM_PID");
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ String appSubmitTimeStr =
+ System.getenv(ApplicationConstants.APP_SUBMIT_TIME_ENV);
+ String clientVersion = System.getenv(TezConstants.TEZ_CLIENT_VERSION_ENV);
+ if (clientVersion == null) {
+ clientVersion = VersionInfo.UNKNOWN;
+ }
+
+ Objects.requireNonNull(appSubmitTimeStr,
+ ApplicationConstants.APP_SUBMIT_TIME_ENV + " is null");
+
+ ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
+ ApplicationAttemptId applicationAttemptId =
+ containerId.getApplicationAttemptId();
+
Review Comment:
nit: remove this blank line.
##########
client-tez/src/main/java/org/apache/tez/dag/app/RssDAGAppMaster.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.dag.app;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.Options;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.mapreduce.JobSubmissionFiles;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ShutdownHookManager;
+import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.SystemClock;
+import org.apache.tez.common.RssTezConfig;
+import org.apache.tez.common.RssTezUtils;
+import org.apache.tez.common.TezClassLoader;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtilsInternal;
+import org.apache.tez.common.VersionInfo;
+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.records.DAGProtos;
+import org.apache.tez.dag.api.records.DAGProtos.AMPluginDescriptorProto;
+import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+
+import static org.apache.tez.common.TezCommonUtils.TEZ_SYSTEM_SUB_DIR;
+
+public class RssDAGAppMaster extends DAGAppMaster {
+ private static final Logger LOG = LoggerFactory.getLogger(RssDAGAppMaster.class);
+ private ShuffleWriteClient shuffleWriteClient;
+ private TezRemoteShuffleManager tezRemoteShuffleManager;
+ private final String rssConfFileLocalResourceName = "rss_conf.xml";
+ private DAGProtos.PlanLocalResource rssConfFileLocalResource;
+ final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ return t;
+ }
+ }
+ );
+
+ public RssDAGAppMaster(ApplicationAttemptId applicationAttemptId,
+ ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
+ Clock clock, long appSubmitTime, boolean isSession, String workingDirectory,
+ String [] localDirs, String[] logDirs, String clientVersion,
+ Credentials credentials, String jobUserName, AMPluginDescriptorProto pluginDescriptorProto) {
+ super(applicationAttemptId, containerId, nmHost, nmPort, nmHttpPort, clock, appSubmitTime, isSession,
+ workingDirectory, localDirs, logDirs, clientVersion, credentials, jobUserName, pluginDescriptorProto);
+ }
+
+ public ShuffleWriteClient getShuffleWriteClient() {
+ return shuffleWriteClient;
+ }
+
+ public void setShuffleWriteClient(ShuffleWriteClient shuffleWriteClient) {
+ this.shuffleWriteClient = shuffleWriteClient;
+ }
+
+ public TezRemoteShuffleManager getTezRemoteShuffleManager() {
+ return tezRemoteShuffleManager;
+ }
+
+ public void setTezRemoteShuffleManager(TezRemoteShuffleManager tezRemoteShuffleManager) {
+ this.tezRemoteShuffleManager = tezRemoteShuffleManager;
+ }
+
+ /**
+ * Init and Start Rss Client
+ * @param appMaster
+ * @param conf
+ * @param applicationAttemptId
+ * @throws Exception
+ */
+ public static void initAndStartRSSClient(final RssDAGAppMaster appMaster, Configuration conf,
+ ApplicationAttemptId applicationAttemptId) throws Exception {
+
+ ShuffleWriteClient client = RssTezUtils.createShuffleClient(conf);
+ appMaster.setShuffleWriteClient(client);
+
+ String coordinators = conf.get(RssTezConfig.RSS_COORDINATOR_QUORUM);
+ LOG.info("Registering coordinators {}", coordinators);
+ client.registerCoordinators(coordinators);
+
+ String strAppAttemptId = applicationAttemptId.toString();
+ long heartbeatInterval = conf.getLong(RssTezConfig.RSS_HEARTBEAT_INTERVAL,
+ RssTezConfig.RSS_HEARTBEAT_INTERVAL_DEFAULT_VALUE);
+ long heartbeatTimeout = conf.getLong(RssTezConfig.RSS_HEARTBEAT_TIMEOUT, heartbeatInterval / 2);
+ client.registerApplicationInfo(strAppAttemptId, heartbeatTimeout, "user");
+
+ appMaster.scheduledExecutorService.scheduleAtFixedRate(
+ () -> {
+ try {
+ client.sendAppHeartbeat(strAppAttemptId, heartbeatTimeout);
+ LOG.info("Finish send heartbeat to coordinator and servers");
+ } catch (Exception e) {
+ LOG.warn("Fail to send heartbeat to coordinator and servers", e);
+ }
+ },
+ heartbeatInterval / 2,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+
+ appMaster.setTezRemoteShuffleManager(new TezRemoteShuffleManager(strAppAttemptId, null, conf,
+ strAppAttemptId, client));
+ appMaster.getTezRemoteShuffleManager().initialize();
+ appMaster.getTezRemoteShuffleManager().start();
+
+ TezConfiguration extraConf = new TezConfiguration(false);
+ extraConf.clear();
+
+ String strAppId = applicationAttemptId.getApplicationId().toString();
+ extraConf.set(RssTezConfig.RSS_AM_SHUFFLE_MANAGER_ADDRESS,
+ appMaster.getTezRemoteShuffleManager().address.getHostName());
+ extraConf.setInt(RssTezConfig.RSS_AM_SHUFFLE_MANAGER_PORT,
+ appMaster.getTezRemoteShuffleManager().address.getPort());
+ writeExtraConf(appMaster, conf, extraConf, strAppId);
+
+ mayCloseTezSlowStart(conf);
+ }
+
+ @Override
+ public String submitDAGToAppMaster(DAGProtos.DAGPlan dagPlan,
+ Map<String, LocalResource> additionalResources) throws TezException {
+ List<DAGProtos.PlanLocalResource> planLocalResourceList = dagPlan.getLocalResourceList();
+
+ if (planLocalResourceList == null) {
+ LOG.warn("planLocalResourceList is null, add new list");
+ planLocalResourceList = new ArrayList<>();
+ } else {
+ planLocalResourceList = new ArrayList<>(planLocalResourceList);
+ }
+
+ try {
+ planLocalResourceList.add(rssConfFileLocalResource);
+ Field field = DAGProtos.DAGPlan.class.getDeclaredField("localResource_");
+ field.setAccessible(true);
+ field.set(dagPlan, planLocalResourceList);
+ field.setAccessible(false);
+ } catch (Exception e) {
+ LOG.error("submitDAGToAppMaster reflect error", e);
+ throw new TezException(e.getMessage());
+ }
+
+ if (LOG.isDebugEnabled()) {
+ for (DAGProtos.PlanLocalResource localResource : dagPlan.getLocalResourceList()) {
+ LOG.debug("localResource: {}", localResource.toString());
+ }
+ }
+
+ return super.submitDAGToAppMaster(dagPlan, additionalResources);
+ }
+
+ /**
+ * main method
+ * @param args
+ */
+ public static void main(String[] args) {
+ try {
+ // Install the tez class loader, which can be used add new resources
+ TezClassLoader.setupTezClassLoader();
+ Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
+ final String pid = System.getenv().get("JVM_PID");
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ String appSubmitTimeStr =
+ System.getenv(ApplicationConstants.APP_SUBMIT_TIME_ENV);
+ String clientVersion = System.getenv(TezConstants.TEZ_CLIENT_VERSION_ENV);
+ if (clientVersion == null) {
+ clientVersion = VersionInfo.UNKNOWN;
+ }
+
+ Objects.requireNonNull(appSubmitTimeStr,
+ ApplicationConstants.APP_SUBMIT_TIME_ENV + " is null");
+
+ ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
+ ApplicationAttemptId applicationAttemptId =
+ containerId.getApplicationAttemptId();
+
+
+ String jobUserName = System
+ .getenv(ApplicationConstants.Environment.USER.name());
+
+ // Command line options
+ Options opts = new Options();
+ opts.addOption(TezConstants.TEZ_SESSION_MODE_CLI_OPTION,
+ false, "Run Tez Application Master in Session mode");
+
+ CommandLine cliParser = new GnuParser().parse(opts, args);
+ boolean sessionModeCliOption = cliParser.hasOption(TezConstants.TEZ_SESSION_MODE_CLI_OPTION);
+
+ LOG.info("Creating DAGAppMaster for "
+ + "applicationId=" + applicationAttemptId.getApplicationId()
+ + ", attemptNum=" + applicationAttemptId.getAttemptId()
+ + ", AMContainerId=" + containerId
+ + ", jvmPid=" + pid
+ + ", userFromEnv=" + jobUserName
+ + ", cliSessionOption=" + sessionModeCliOption
+ + ", pwd=" + System.getenv(ApplicationConstants.Environment.PWD.name())
+ + ", localDirs=" + System.getenv(ApplicationConstants.Environment.LOCAL_DIRS.name())
+ + ", logDirs=" + System.getenv(ApplicationConstants.Environment.LOG_DIRS.name()));
+
+ Configuration conf = new Configuration(new YarnConfiguration());
Review Comment:
Why we need `YarnConfiguration` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org