You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "lifeSo (via GitHub)" <gi...@apache.org> on 2023/05/17 10:00:04 UTC
[GitHub] [incubator-uniffle] lifeSo opened a new pull request, #890: Common
lifeSo opened a new pull request, #890:
URL: https://github.com/apache/incubator-uniffle/pull/890
<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
Examples:
- "[#123] feat(operator): support xxx"
- "[#233] fix: check null before access result in xxx"
- "[MINOR] refactor: fix typo in variable name"
- "[MINOR] docs: fix typo in README"
- "[#255] test: fix flaky test NameOfTheTest"
Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. Contributor guidelines:
https://github.com/apache/incubator-uniffle/blob/master/CONTRIBUTING.md
3. If the PR is unfinished, please mark this PR as draft.
-->
### What changes were proposed in this pull request?
Tez Shuffle Read supporting Remote Shuffle related common and utils
### Why are the changes needed?
Fix: https://github.com/apache/incubator-uniffle/issues/872
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
We will add test cases in a few days
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198644000
##########
client-tez/pom.xml:
##########
@@ -26,61 +27,81 @@
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>org.apache.uniffle</groupId>
<artifactId>rss-client-tez</artifactId>
<version>0.8.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Apache Uniffle Client (Tez)</name>
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
Review Comment:
ok, we removed it and pushed it.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197928528
##########
client-tez/src/test/java/org/apache/tez/common/TezIdHelperTest.java:
##########
@@ -0,0 +1,17 @@
+package org.apache.tez.common;
+
+import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TezIdHelperTest {
+
+ @Test
+ public void TestTetTaskAttemptId(){
+ TezIdHelper tezIdHelper = new TezIdHelper();
Review Comment:
ok and done
--
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
[GitHub] [incubator-uniffle] zhengchenyu commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198513685
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String PLUS_DELIMITER = "+";
+ public static final String UNDERLINE_DELIMITER = "_";
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ // constant to compute shuffle id
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+ private static final int VERTEX_ID_MAPPING_MAGIC = 600;
+ private static final int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ private RssTezUtils() {
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+ LOG.info("InitialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("MaxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getNettyPort();
+ res.add(tmp);
+ }
+ }
+ return org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+ }
+
+ public static Map<String, List<String>> uniformServerToPartitions(String partitionToServers) {
+ Map<String, List<String>> serverToPartitions = new HashMap<>();
+ List<String> list;
+
+ String[] pidWithWorkerInfos = partitionToServers.split(COMMA_DELIMITER);
+ for (String pidWithWorkerInfo : pidWithWorkerInfos) {
+ String[] pidUnderLineWorkerInfo = pidWithWorkerInfo.split(UNDERLINE_DELIMITER);
+ if (serverToPartitions.containsKey(pidUnderLineWorkerInfo[1])) {
+ list = serverToPartitions.get(pidUnderLineWorkerInfo[1]);
+ list.add(pidUnderLineWorkerInfo[0]);
+ } else {
+ list = new ArrayList<>();
+ list.add(pidUnderLineWorkerInfo[0]);
+ serverToPartitions.put(pidUnderLineWorkerInfo[1], list);
+ }
+ }
+
+ return serverToPartitions;
+ }
+
+ public static String uniformServerToPartitions(Map<String, List<String>> map) {
+ List<String> res = new ArrayList<>();
+ Set<String> keySet = map.keySet();
+ for (String s : keySet) {
+ String join = org.apache.commons.lang.StringUtils.join(map.get(s), UNDERLINE_DELIMITER);
+ res.add(s + PLUS_DELIMITER + join);
+ }
+
+ return org.apache.commons.lang.StringUtils.join(res,COMMA_DELIMITER);
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
Review Comment:
1. name fix
rssJobConf -> rssConf, mrJobCOnf -> tezConf
2. In client-mr mode, we use rss_conf.xml to transmit rss config. In client-tez mode, I think rss_conf.xml is unnecessary, so we can retain only one config in parameters.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198480230
##########
client-tez/pom.xml:
##########
@@ -176,4 +181,4 @@
</plugin>
</plugins>
</build>
-</project>
+</project>
Review Comment:
ok, we add added it
--
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
[GitHub] [incubator-uniffle] jerqi merged pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi merged PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198645154
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ <version>${tez.version}</version>
Review Comment:
ok, we removed version config and pushed the code.
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-api</artifactId>
+ <version>${tez.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
-
<dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
Review Comment:
ok, we removed version config and pushed the code.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198668550
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String PLUS_DELIMITER = "+";
+ public static final String UNDERLINE_DELIMITER = "_";
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ // constant to compute shuffle id
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+ private static final int VERTEX_ID_MAPPING_MAGIC = 600;
+ private static final int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ private RssTezUtils() {
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+ LOG.info("InitialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("MaxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getNettyPort();
+ res.add(tmp);
+ }
+ }
+ return org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+ }
+
+ public static Map<String, List<String>> uniformServerToPartitions(String partitionToServers) {
+ Map<String, List<String>> serverToPartitions = new HashMap<>();
+ List<String> list;
+
+ String[] pidWithWorkerInfos = partitionToServers.split(COMMA_DELIMITER);
+ for (String pidWithWorkerInfo : pidWithWorkerInfos) {
+ String[] pidUnderLineWorkerInfo = pidWithWorkerInfo.split(UNDERLINE_DELIMITER);
+ if (serverToPartitions.containsKey(pidUnderLineWorkerInfo[1])) {
+ list = serverToPartitions.get(pidUnderLineWorkerInfo[1]);
+ list.add(pidUnderLineWorkerInfo[0]);
+ } else {
+ list = new ArrayList<>();
+ list.add(pidUnderLineWorkerInfo[0]);
+ serverToPartitions.put(pidUnderLineWorkerInfo[1], list);
+ }
+ }
+
+ return serverToPartitions;
+ }
+
+ public static String uniformServerToPartitions(Map<String, List<String>> map) {
+ List<String> res = new ArrayList<>();
+ Set<String> keySet = map.keySet();
+ for (String s : keySet) {
+ String join = org.apache.commons.lang.StringUtils.join(map.get(s), UNDERLINE_DELIMITER);
+ res.add(s + PLUS_DELIMITER + join);
+ }
+
+ return org.apache.commons.lang.StringUtils.join(res,COMMA_DELIMITER);
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
Review Comment:
It's ok to refactor. rssJobConf -> rssConf, mrJobConf -> tezConf.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198480114
##########
client-tez/pom.xml:
##########
@@ -1,20 +1,4 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!--
Review Comment:
ok, we added it
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1196580354
##########
client-tez/pom.xml:
##########
@@ -1,86 +1,91 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>uniffle-parent</artifactId>
<groupId>org.apache.uniffle</groupId>
- <version>0.8.0-SNAPSHOT</version>
+ <version>0.7.0</version>
Review Comment:
Could you modify the version? It's wrong.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198519246
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String PLUS_DELIMITER = "+";
+ public static final String UNDERLINE_DELIMITER = "_";
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ // constant to compute shuffle id
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+ private static final int VERTEX_ID_MAPPING_MAGIC = 600;
+ private static final int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ private RssTezUtils() {
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+ LOG.info("InitialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("MaxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getNettyPort();
+ res.add(tmp);
+ }
+ }
+ return org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+ }
+
+ public static Map<String, List<String>> uniformServerToPartitions(String partitionToServers) {
+ Map<String, List<String>> serverToPartitions = new HashMap<>();
+ List<String> list;
+
+ String[] pidWithWorkerInfos = partitionToServers.split(COMMA_DELIMITER);
+ for (String pidWithWorkerInfo : pidWithWorkerInfos) {
+ String[] pidUnderLineWorkerInfo = pidWithWorkerInfo.split(UNDERLINE_DELIMITER);
+ if (serverToPartitions.containsKey(pidUnderLineWorkerInfo[1])) {
+ list = serverToPartitions.get(pidUnderLineWorkerInfo[1]);
+ list.add(pidUnderLineWorkerInfo[0]);
+ } else {
+ list = new ArrayList<>();
+ list.add(pidUnderLineWorkerInfo[0]);
+ serverToPartitions.put(pidUnderLineWorkerInfo[1], list);
+ }
+ }
+
+ return serverToPartitions;
+ }
+
+ public static String uniformServerToPartitions(Map<String, List<String>> map) {
+ List<String> res = new ArrayList<>();
+ Set<String> keySet = map.keySet();
+ for (String s : keySet) {
+ String join = org.apache.commons.lang.StringUtils.join(map.get(s), UNDERLINE_DELIMITER);
+ res.add(s + PLUS_DELIMITER + join);
+ }
+
+ return org.apache.commons.lang.StringUtils.join(res,COMMA_DELIMITER);
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
Review Comment:
2. we have dynamic configuration mechnism, Coordinator will change the config option of the client's configuration. We need another configuration to save changed config option.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198485851
##########
client-tez/pom.xml:
##########
@@ -26,61 +27,81 @@
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>org.apache.uniffle</groupId>
<artifactId>rss-client-tez</artifactId>
<version>0.8.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Apache Uniffle Client (Tez)</name>
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
Review Comment:
Could we remove the `version`? The pom will extend the parent tez version. We use the parent pom to manage the third party package version.
--
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
[GitHub] [incubator-uniffle] jerqi commented on pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#issuecomment-1551174939
This pr is also a little large. Could you split into more smaller prs?
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1196311846
##########
client-tez/src/test/java/org/apache/tez/common/TezIdHelperTest.java:
##########
@@ -0,0 +1,17 @@
+package org.apache.tez.common;
+
+import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TezIdHelperTest {
+
+ @Test
+ public void TestTetTaskAttemptId(){
+ TezIdHelper tezIdHelper = new TezIdHelper();
Review Comment:
Could we use assert to judge whether the value is correct?
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1196313830
##########
client-tez/src/main/java/org/apache/tez/common/GetShuffleServerResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Objects;
+
+import org.apache.hadoop.io.Writable;
+
+public class GetShuffleServerResponse implements Writable {
Review Comment:
Could we remove this class in this pr? According to offline discussion, we may use another plan.
##########
client-tez/src/main/java/org/apache/tez/common/TezClassLoader.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.common;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * ClassLoader to allow addition of new paths to classpath in the runtime.
+ *
+ * It uses URLClassLoader with this class' classloader as parent classloader.
+ * And hence first delegates the resource loading to parent and then to the URLs
+ * added. The process must be setup to use by invoking setupTezClassLoader() which sets
+ * the global TezClassLoader as current thread context class loader. All threads
+ * created will inherit the classloader and hence will resolve the class/resource
+ * from TezClassLoader.
+ */
+
+public class TezClassLoader extends URLClassLoader {
Review Comment:
Why do need this class? Could we reuse Tez's implement?
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
+ int upVertexId = mapVertexId(upVertexName);
+ int downVertexId = mapVertexId(downVertexName);
+ int shuffleId = computeShuffleId(tezDagID, upVertexId, downVertexId);
+ LOG.info("Compute Shuffle Id, upVertexName:{}, id:{}, downVertexName:{}, id:{}, shuffleId:{}",
+ upVertexName, upVertexId, downVertexName, downVertexId, shuffleId);
+ return shuffleId;
+ }
+
+ public static int computeShuffleId(int tezDagID, int upTezVertexID, int downTezVertexID) {
+ return tezDagID * (SHUFFLE_ID_MAGIC * SHUFFLE_ID_MAGIC) + upTezVertexID * SHUFFLE_ID_MAGIC + downTezVertexID;
+ }
+
+ public static int computeShuffleId(int tezDagID, int tezVertexID) {
+ return tezDagID * 10000 + tezVertexID;
+ }
+
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+
+ private static int mapVertexId(String vertexName) {
+ String[] ss = vertexName.split("\\s+");
+ if (Integer.parseInt(ss[1]) > VERTEX_ID_MAPPING_MAX_ID) {
+ throw new RssException("Too large vertex name to id mapping, vertexName:" + vertexName);
+ }
+ if (VERTEX_ID_MAPPING_MAP.equals(ss[0])) {
+ return Integer.parseInt(ss[1]);
+ } else if (VERTEX_ID_MAPPING_REDUCER.equals(ss[0])) {
+ return 600 + Integer.parseInt(ss[1]);
+ } else {
+ throw new RssException("Wrong vertex name to id mapping, vertexName:" + vertexName);
+ }
+ }
+
+
+
+ private static int SHUFFLE_ID_MAGIC = 1000;
Review Comment:
Could we use `final` key word herer?
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
+ int upVertexId = mapVertexId(upVertexName);
+ int downVertexId = mapVertexId(downVertexName);
+ int shuffleId = computeShuffleId(tezDagID, upVertexId, downVertexId);
+ LOG.info("Compute Shuffle Id, upVertexName:{}, id:{}, downVertexName:{}, id:{}, shuffleId:{}",
+ upVertexName, upVertexId, downVertexName, downVertexId, shuffleId);
+ return shuffleId;
+ }
+
+ public static int computeShuffleId(int tezDagID, int upTezVertexID, int downTezVertexID) {
+ return tezDagID * (SHUFFLE_ID_MAGIC * SHUFFLE_ID_MAGIC) + upTezVertexID * SHUFFLE_ID_MAGIC + downTezVertexID;
+ }
+
+ public static int computeShuffleId(int tezDagID, int tezVertexID) {
+ return tezDagID * 10000 + tezVertexID;
+ }
+
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
Review Comment:
Could we put the constants in the top place of this class?
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
Review Comment:
Could you add some comments to explain how to generate the shuffleID according to vertex information?
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
+ int upVertexId = mapVertexId(upVertexName);
+ int downVertexId = mapVertexId(downVertexName);
+ int shuffleId = computeShuffleId(tezDagID, upVertexId, downVertexId);
+ LOG.info("Compute Shuffle Id, upVertexName:{}, id:{}, downVertexName:{}, id:{}, shuffleId:{}",
+ upVertexName, upVertexId, downVertexName, downVertexId, shuffleId);
+ return shuffleId;
+ }
+
+ public static int computeShuffleId(int tezDagID, int upTezVertexID, int downTezVertexID) {
+ return tezDagID * (SHUFFLE_ID_MAGIC * SHUFFLE_ID_MAGIC) + upTezVertexID * SHUFFLE_ID_MAGIC + downTezVertexID;
+ }
+
+ public static int computeShuffleId(int tezDagID, int tezVertexID) {
+ return tezDagID * 10000 + tezVertexID;
+ }
+
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+
+ private static int mapVertexId(String vertexName) {
+ String[] ss = vertexName.split("\\s+");
+ if (Integer.parseInt(ss[1]) > VERTEX_ID_MAPPING_MAX_ID) {
+ throw new RssException("Too large vertex name to id mapping, vertexName:" + vertexName);
+ }
+ if (VERTEX_ID_MAPPING_MAP.equals(ss[0])) {
+ return Integer.parseInt(ss[1]);
+ } else if (VERTEX_ID_MAPPING_REDUCER.equals(ss[0])) {
+ return 600 + Integer.parseInt(ss[1]);
+ } else {
+ throw new RssException("Wrong vertex name to id mapping, vertexName:" + vertexName);
+ }
+ }
+
+
+
+ private static int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ public static int computeVertexId(InputAttemptIdentifier inputAttemptIdentifier) {
+ return TezTaskAttemptID.fromString(inputAttemptIdentifier.getPathComponent()).getTaskID().getVertexID().getId();
+ }
+
+ public static long convertTaskAttemptIdToLong(TezTaskAttemptID taskAttemptID, int appAttemptId) {
+ long lowBytes = taskAttemptID.getTaskID().getId();
+ if (lowBytes > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " low bytes " + lowBytes + " exceed");
+ }
+ if (appAttemptId < 1) {
+ throw new RssException("appAttemptId " + appAttemptId + " is wrong");
+ }
+ long highBytes = (long)taskAttemptID.getId() - (appAttemptId - 1) * 1000;
+ if (highBytes > MAX_ATTEMPT_ID || highBytes < 0) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " high bytes " + highBytes + " exceed");
+ }
+
+ long id = (highBytes << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + lowBytes;
+ LOG.info("convertTaskAttemptIdToLong id is {}", id);
+ LOG.info("lowBytes id is {}", lowBytes);
+ LOG.info("highBytes id is {}", highBytes);
+ return id;
+ }
+
+ public static long convertTaskAttemptIdToLong(int taskId, int taskAttemptID, int appAttemptId) {
+ long lowBytes = taskId;
+ if (lowBytes > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " low bytes " + lowBytes + " exceed");
+ }
+ if (appAttemptId < 1) {
+ throw new RssException("appAttemptId " + appAttemptId + " is wrong");
+ }
+ long highBytes = (long)taskAttemptID - (appAttemptId - 1) * 1000;
+ if (highBytes > MAX_ATTEMPT_ID || highBytes < 0) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " high bytes " + highBytes + " exceed");
+ }
+ return (highBytes << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + lowBytes;
+ }
+
+ public static Roaring64NavigableMap fetchAllRssTaskIds(Set<InputAttemptIdentifier> successMapTaskAttempts,
Review Comment:
```
public static Roaring64NavigableMap fetchAllRssTaskIds(
Set<InputAttemptIdentifier> successMapTaskAttempts,
Integer totalMapsCount,
Integer appAttemptId
````
Could we keep the same style as other places?
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197929666
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
+ int upVertexId = mapVertexId(upVertexName);
+ int downVertexId = mapVertexId(downVertexName);
+ int shuffleId = computeShuffleId(tezDagID, upVertexId, downVertexId);
+ LOG.info("Compute Shuffle Id, upVertexName:{}, id:{}, downVertexName:{}, id:{}, shuffleId:{}",
+ upVertexName, upVertexId, downVertexName, downVertexId, shuffleId);
+ return shuffleId;
+ }
+
+ public static int computeShuffleId(int tezDagID, int upTezVertexID, int downTezVertexID) {
+ return tezDagID * (SHUFFLE_ID_MAGIC * SHUFFLE_ID_MAGIC) + upTezVertexID * SHUFFLE_ID_MAGIC + downTezVertexID;
+ }
+
+ public static int computeShuffleId(int tezDagID, int tezVertexID) {
+ return tezDagID * 10000 + tezVertexID;
+ }
+
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+
+ private static int mapVertexId(String vertexName) {
+ String[] ss = vertexName.split("\\s+");
+ if (Integer.parseInt(ss[1]) > VERTEX_ID_MAPPING_MAX_ID) {
+ throw new RssException("Too large vertex name to id mapping, vertexName:" + vertexName);
+ }
+ if (VERTEX_ID_MAPPING_MAP.equals(ss[0])) {
+ return Integer.parseInt(ss[1]);
+ } else if (VERTEX_ID_MAPPING_REDUCER.equals(ss[0])) {
+ return 600 + Integer.parseInt(ss[1]);
+ } else {
+ throw new RssException("Wrong vertex name to id mapping, vertexName:" + vertexName);
+ }
+ }
+
+
+
+ private static int SHUFFLE_ID_MAGIC = 1000;
Review Comment:
ok and corrected it
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198511521
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
Review Comment:
ditto.
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
Review Comment:
ditto.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198645638
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-api</artifactId>
+ <version>${tez.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
-
<dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
</dependency>
<dependency>
- <groupId>io.grpc</groupId>
- <artifactId>grpc-netty-shaded</artifactId>
- <version>${grpc.version}</version>
+ <groupId>org.apache.uniffle</groupId>
+ <artifactId>rss-client</artifactId>
</dependency>
<dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-all</artifactId>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
- <groupId>io.grpc</groupId>
- <artifactId>grpc-protobuf</artifactId>
+ <groupId>net.jpountz.lz4</groupId>
+ <artifactId>lz4</artifactId>
</dependency>
+
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>${protobuf.version}</version>
</dependency>
+
<dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>com.github.luben</groupId>
- <artifactId>zstd-jni</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${hadoop.version}</version>
Review Comment:
ok, we removed version config and pushed the code.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197300701
##########
pom.xml:
##########
@@ -1625,10 +1625,13 @@
</dependencyManagement>
</profile>
<profile>
- <id>tez</id>
+ <id>tez0.9</id>
Review Comment:
If we only have one version Tez, we don't need the name tez + version. We can use tez directly.
--
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
[GitHub] [incubator-uniffle] jerqi commented on pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#issuecomment-1553923089
cc @zhengchenyu Do you have another suggestion?
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198519246
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String PLUS_DELIMITER = "+";
+ public static final String UNDERLINE_DELIMITER = "_";
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ // constant to compute shuffle id
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+ private static final int VERTEX_ID_MAPPING_MAGIC = 600;
+ private static final int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ private RssTezUtils() {
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+ LOG.info("InitialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("MaxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getNettyPort();
+ res.add(tmp);
+ }
+ }
+ return org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+ }
+
+ public static Map<String, List<String>> uniformServerToPartitions(String partitionToServers) {
+ Map<String, List<String>> serverToPartitions = new HashMap<>();
+ List<String> list;
+
+ String[] pidWithWorkerInfos = partitionToServers.split(COMMA_DELIMITER);
+ for (String pidWithWorkerInfo : pidWithWorkerInfos) {
+ String[] pidUnderLineWorkerInfo = pidWithWorkerInfo.split(UNDERLINE_DELIMITER);
+ if (serverToPartitions.containsKey(pidUnderLineWorkerInfo[1])) {
+ list = serverToPartitions.get(pidUnderLineWorkerInfo[1]);
+ list.add(pidUnderLineWorkerInfo[0]);
+ } else {
+ list = new ArrayList<>();
+ list.add(pidUnderLineWorkerInfo[0]);
+ serverToPartitions.put(pidUnderLineWorkerInfo[1], list);
+ }
+ }
+
+ return serverToPartitions;
+ }
+
+ public static String uniformServerToPartitions(Map<String, List<String>> map) {
+ List<String> res = new ArrayList<>();
+ Set<String> keySet = map.keySet();
+ for (String s : keySet) {
+ String join = org.apache.commons.lang.StringUtils.join(map.get(s), UNDERLINE_DELIMITER);
+ res.add(s + PLUS_DELIMITER + join);
+ }
+
+ return org.apache.commons.lang.StringUtils.join(res,COMMA_DELIMITER);
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
Review Comment:
2. we have dynamic configuration mechnism, Coordinator will change the config option of the client's configuration.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197992473
##########
client-tez/pom.xml:
##########
@@ -1,20 +1,4 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!--
Review Comment:
Why do you remove the license? we must include the Apache License in every file.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198644967
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
Review Comment:
ok, we removed version config and pushed the code.
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
Review Comment:
ok, we removed version config and pushed the code.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198655580
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String PLUS_DELIMITER = "+";
+ public static final String UNDERLINE_DELIMITER = "_";
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ // constant to compute shuffle id
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+ private static final int VERTEX_ID_MAPPING_MAGIC = 600;
+ private static final int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ private RssTezUtils() {
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+ LOG.info("InitialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("MaxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getNettyPort();
+ res.add(tmp);
+ }
+ }
+ return org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+ }
+
+ public static Map<String, List<String>> uniformServerToPartitions(String partitionToServers) {
+ Map<String, List<String>> serverToPartitions = new HashMap<>();
+ List<String> list;
+
+ String[] pidWithWorkerInfos = partitionToServers.split(COMMA_DELIMITER);
+ for (String pidWithWorkerInfo : pidWithWorkerInfos) {
+ String[] pidUnderLineWorkerInfo = pidWithWorkerInfo.split(UNDERLINE_DELIMITER);
+ if (serverToPartitions.containsKey(pidUnderLineWorkerInfo[1])) {
+ list = serverToPartitions.get(pidUnderLineWorkerInfo[1]);
+ list.add(pidUnderLineWorkerInfo[0]);
+ } else {
+ list = new ArrayList<>();
+ list.add(pidUnderLineWorkerInfo[0]);
+ serverToPartitions.put(pidUnderLineWorkerInfo[1], list);
+ }
+ }
+
+ return serverToPartitions;
+ }
+
+ public static String uniformServerToPartitions(Map<String, List<String>> map) {
+ List<String> res = new ArrayList<>();
+ Set<String> keySet = map.keySet();
+ for (String s : keySet) {
+ String join = org.apache.commons.lang.StringUtils.join(map.get(s), UNDERLINE_DELIMITER);
+ res.add(s + PLUS_DELIMITER + join);
+ }
+
+ return org.apache.commons.lang.StringUtils.join(res,COMMA_DELIMITER);
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
Review Comment:
ok, we removeed the getInt method()
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197780407
##########
client-tez/src/main/java/org/apache/tez/common/TezClassLoader.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.common;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * ClassLoader to allow addition of new paths to classpath in the runtime.
+ *
+ * It uses URLClassLoader with this class' classloader as parent classloader.
+ * And hence first delegates the resource loading to parent and then to the URLs
+ * added. The process must be setup to use by invoking setupTezClassLoader() which sets
+ * the global TezClassLoader as current thread context class loader. All threads
+ * created will inherit the classloader and hence will resolve the class/resource
+ * from TezClassLoader.
+ */
+
+public class TezClassLoader extends URLClassLoader {
Review Comment:
Install the tez class loader, which can be used add new resources
click below to refer more:
https://github.com/apache/tez/blob/ccad71dabceb850b58ff38e77fd206e68f62ce4b/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java#L2338
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198480344
##########
client-tez/src/test/java/org/apache/tez/common/TezIdHelperTest.java:
##########
@@ -0,0 +1,16 @@
+package org.apache.tez.common;
Review Comment:
ok, we added it
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198505731
##########
client-tez/pom.xml:
##########
@@ -26,61 +27,81 @@
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>org.apache.uniffle</groupId>
Review Comment:
ok, we reverted it.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198511623
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ <version>${tez.version}</version>
Review Comment:
ditto.
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-api</artifactId>
+ <version>${tez.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
-
<dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
Review Comment:
ditto.
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197300701
##########
pom.xml:
##########
@@ -1625,10 +1625,13 @@
</dependencyManagement>
</profile>
<profile>
- <id>tez</id>
+ <id>tez0.9</id>
Review Comment:
If we only have one version Tez, we don't need the name tez + version.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197927770
##########
client-tez/src/test/java/org/apache/tez/common/RssTezUtilsTest.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.common;
+
+import com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.apache.uniffle.client.util.RssClientConfig;
+import org.apache.uniffle.common.util.Constants;
+import org.apache.uniffle.storage.util.StorageType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class RssTezUtilsTest {
+
+ private static final String EXPECTED_EXCEPTION_MESSAGE = "Exception should be thrown";
+
+// @Test
Review Comment:
ok
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198486186
##########
client-tez/pom.xml:
##########
@@ -26,61 +27,81 @@
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>org.apache.uniffle</groupId>
Review Comment:
Why do we remove this groupId? Could we keep consistent style with other submodule pom files?
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1196306562
##########
client-tez/src/test/java/org/apache/tez/common/RssTezUtilsTest.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.common;
+
+import com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.apache.uniffle.client.util.RssClientConfig;
+import org.apache.uniffle.common.util.Constants;
+import org.apache.uniffle.storage.util.StorageType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class RssTezUtilsTest {
+
+ private static final String EXPECTED_EXCEPTION_MESSAGE = "Exception should be thrown";
+
+// @Test
Review Comment:
Could we remove this unused comment code?
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197991129
##########
client-tez/pom.xml:
##########
@@ -176,4 +181,4 @@
</plugin>
</plugins>
</build>
-</project>
+</project>
Review Comment:
We should add a new line, otherwise there will be a warning.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198476549
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
Review Comment:
Ok, we added it and commit it.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197929057
##########
client-tez/src/main/java/org/apache/tez/common/GetShuffleServerResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Objects;
+
+import org.apache.hadoop.io.Writable;
+
+public class GetShuffleServerResponse implements Writable {
Review Comment:
ok, we removed the file in this commit
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
+ int upVertexId = mapVertexId(upVertexName);
+ int downVertexId = mapVertexId(downVertexName);
+ int shuffleId = computeShuffleId(tezDagID, upVertexId, downVertexId);
+ LOG.info("Compute Shuffle Id, upVertexName:{}, id:{}, downVertexName:{}, id:{}, shuffleId:{}",
+ upVertexName, upVertexId, downVertexName, downVertexId, shuffleId);
+ return shuffleId;
+ }
+
+ public static int computeShuffleId(int tezDagID, int upTezVertexID, int downTezVertexID) {
+ return tezDagID * (SHUFFLE_ID_MAGIC * SHUFFLE_ID_MAGIC) + upTezVertexID * SHUFFLE_ID_MAGIC + downTezVertexID;
+ }
+
+ public static int computeShuffleId(int tezDagID, int tezVertexID) {
+ return tezDagID * 10000 + tezVertexID;
+ }
+
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
Review Comment:
ok and changed
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198483444
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,470 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String UNDERLINE_DELIMITER = "_";
+
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ private RssTezUtils(){
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+
+ LOG.info("initialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("maxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getPort();
+ res.add(tmp);
+ }
+ }
+ String str = org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+
+ return str;
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr =
+ System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+
+
+ public static void applyDynamicClientConf(Configuration jobConf, Map<String, String> confItems) {
+ if (jobConf == null) {
+ LOG.warn("Job conf is null");
+ return;
+ }
+
+ if (confItems == null || confItems.isEmpty()) {
+ LOG.warn("Empty conf items");
+ return;
+ }
+
+ for (Map.Entry<String, String> kv : confItems.entrySet()) {
+ String mrConfKey = kv.getKey();
+ if (!mrConfKey.startsWith(RssTezConfig.TEZ_RSS_CONFIG_PREFIX)) {
+ mrConfKey = RssTezConfig.TEZ_RSS_CONFIG_PREFIX + mrConfKey;
+ }
+ String mrConfVal = kv.getValue();
+ if (StringUtils.isEmpty(jobConf.get(mrConfKey, ""))
+ || RssTezConfig.RSS_MANDATORY_CLUSTER_CONF.contains(mrConfKey)) {
+ LOG.warn("Use conf dynamic conf {} = {}", mrConfKey, mrConfVal);
+ jobConf.set(mrConfKey, mrConfVal);
+ }
+ }
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
+ return rssJobConf.getInt(key, mrJobCOnf.getInt(key, defaultValue));
+ }
+
+ public static long getLong(Configuration rssJobConf, Configuration mrJobConf, String key, long defaultValue) {
+ return rssJobConf.getLong(key, mrJobConf.getLong(key, defaultValue));
+ }
+
+ public static boolean getBoolean(Configuration rssJobConf, Configuration mrJobConf, String key,
+ boolean defaultValue) {
+ return rssJobConf.getBoolean(key, mrJobConf.getBoolean(key, defaultValue));
+ }
+
+ public static double getDouble(Configuration rssJobConf, Configuration mrJobConf, String key, double defaultValue) {
+ return rssJobConf.getDouble(key, mrJobConf.getDouble(key, defaultValue));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key) {
+ return rssJobConf.get(key, mrJobConf.get(key));
+ }
+
+ public static String getString(Configuration rssJobConf, Configuration mrJobConf, String key, String defaultValue) {
+ return rssJobConf.get(key, mrJobConf.get(key, defaultValue));
+ }
+
+ public static long getBlockId(long partitionId, long taskAttemptId, int nextSeqNo) {
+ long attemptId = taskAttemptId >> (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH);
+ if (attemptId < 0 || attemptId > MAX_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support attemptId [" + attemptId
+ + "], the max value should be " + MAX_ATTEMPT_ID);
+ }
+ long atomicInt = (nextSeqNo << MAX_ATTEMPT_LENGTH) + attemptId;
+ if (atomicInt < 0 || atomicInt > Constants.MAX_SEQUENCE_NO) {
+ throw new RuntimeException("Can't support sequence [" + atomicInt
+ + "], the max value should be " + Constants.MAX_SEQUENCE_NO);
+ }
+ if (partitionId < 0 || partitionId > Constants.MAX_PARTITION_ID) {
+ throw new RuntimeException("Can't support partitionId["
+ + partitionId + "], the max value should be " + Constants.MAX_PARTITION_ID);
+ }
+ long taskId = taskAttemptId - (attemptId
+ << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH));
+ if (taskId < 0 || taskId > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RuntimeException("Can't support taskId["
+ + taskId + "], the max value should be " + Constants.MAX_TASK_ATTEMPT_ID);
+ }
+ return (atomicInt << (Constants.PARTITION_ID_MAX_LENGTH + Constants.TASK_ATTEMPT_ID_MAX_LENGTH))
+ + (partitionId << Constants.TASK_ATTEMPT_ID_MAX_LENGTH) + taskId;
+ }
+
+ public static long getTaskAttemptId(long blockId) {
+ long mapId = blockId & Constants.MAX_TASK_ATTEMPT_ID;
+ long attemptId = (blockId >> (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH))
+ & MAX_ATTEMPT_ID;
+ return (attemptId << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + mapId;
+ }
+
+ public static int estimateTaskConcurrency(Configuration jobConf, int mapNum, int reduceNum) {
+ double dynamicFactor = jobConf.getDouble(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_DYNAMIC_FACTOR_DEFAULT_VALUE);
+ double slowStart = jobConf.getDouble(Constants.MR_SLOW_START, Constants.MR_SLOW_START_DEFAULT_VALUE);
+ int mapLimit = jobConf.getInt(Constants.MR_MAP_LIMIT, Constants.MR_MAP_LIMIT_DEFAULT_VALUE);
+ int reduceLimit = jobConf.getInt(Constants.MR_REDUCE_LIMIT, Constants.MR_REDUCE_LIMIT_DEFAULT_VALUE);
+
+ int estimateMapNum = mapLimit > 0 ? Math.min(mapNum, mapLimit) : mapNum;
+ int estimateReduceNum = reduceLimit > 0 ? Math.min(reduceNum, reduceLimit) : reduceNum;
+ if (slowStart == 1) {
+ return (int) (Math.max(estimateMapNum, estimateReduceNum) * dynamicFactor);
+ } else {
+ return (int) (((1 - slowStart) * estimateMapNum + estimateReduceNum) * dynamicFactor);
+ }
+ }
+
+ public static int getRequiredShuffleServerNumber(Configuration jobConf, int mapNum, int reduceNum) {
+ int requiredShuffleServerNumber = jobConf.getInt(
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER,
+ RssTezConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER_DEFAULT_VALUE
+ );
+ boolean enabledEstimateServer = jobConf.getBoolean(
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED,
+ RssTezConfig.RSS_ESTIMATE_SERVER_ASSIGNMENT_ENABLED_DEFAULT_VALUE
+ );
+ if (!enabledEstimateServer || requiredShuffleServerNumber > 0) {
+ return requiredShuffleServerNumber;
+ }
+ int taskConcurrency = estimateTaskConcurrency(jobConf, mapNum, reduceNum);
+ int taskConcurrencyPerServer = jobConf.getInt(RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER,
+ RssTezConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ return (int) Math.ceil(taskConcurrency * 1.0 / taskConcurrencyPerServer);
+ }
+
+ public static void validateRssClientConf(Configuration rssJobConf, Configuration mrJobConf) {
+ int retryMax = getInt(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ long sendCheckTimeout = getLong(rssJobConf, mrJobConf, RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS_DEFAULT_VALUE);
+ if (retryIntervalMax * retryMax > sendCheckTimeout) {
+ throw new IllegalArgumentException(String.format("%s(%s) * %s(%s) should not bigger than %s(%s)",
+ RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ retryMax,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ retryIntervalMax,
+ RssTezConfig.RSS_CLIENT_SEND_CHECK_TIMEOUT_MS,
+ sendCheckTimeout));
+ }
+ }
+
+ public static int computeShuffleId(int tezDagID, String upVertexName, String downVertexName) {
+ int upVertexId = mapVertexId(upVertexName);
+ int downVertexId = mapVertexId(downVertexName);
+ int shuffleId = computeShuffleId(tezDagID, upVertexId, downVertexId);
+ LOG.info("Compute Shuffle Id, upVertexName:{}, id:{}, downVertexName:{}, id:{}, shuffleId:{}",
+ upVertexName, upVertexId, downVertexName, downVertexId, shuffleId);
+ return shuffleId;
+ }
+
+ public static int computeShuffleId(int tezDagID, int upTezVertexID, int downTezVertexID) {
+ return tezDagID * (SHUFFLE_ID_MAGIC * SHUFFLE_ID_MAGIC) + upTezVertexID * SHUFFLE_ID_MAGIC + downTezVertexID;
+ }
+
+ public static int computeShuffleId(int tezDagID, int tezVertexID) {
+ return tezDagID * 10000 + tezVertexID;
+ }
+
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+
+ private static int mapVertexId(String vertexName) {
+ String[] ss = vertexName.split("\\s+");
+ if (Integer.parseInt(ss[1]) > VERTEX_ID_MAPPING_MAX_ID) {
+ throw new RssException("Too large vertex name to id mapping, vertexName:" + vertexName);
+ }
+ if (VERTEX_ID_MAPPING_MAP.equals(ss[0])) {
+ return Integer.parseInt(ss[1]);
+ } else if (VERTEX_ID_MAPPING_REDUCER.equals(ss[0])) {
+ return 600 + Integer.parseInt(ss[1]);
+ } else {
+ throw new RssException("Wrong vertex name to id mapping, vertexName:" + vertexName);
+ }
+ }
+
+
+
+ private static int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ public static int computeVertexId(InputAttemptIdentifier inputAttemptIdentifier) {
+ return TezTaskAttemptID.fromString(inputAttemptIdentifier.getPathComponent()).getTaskID().getVertexID().getId();
+ }
+
+ public static long convertTaskAttemptIdToLong(TezTaskAttemptID taskAttemptID, int appAttemptId) {
+ long lowBytes = taskAttemptID.getTaskID().getId();
+ if (lowBytes > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " low bytes " + lowBytes + " exceed");
+ }
+ if (appAttemptId < 1) {
+ throw new RssException("appAttemptId " + appAttemptId + " is wrong");
+ }
+ long highBytes = (long)taskAttemptID.getId() - (appAttemptId - 1) * 1000;
+ if (highBytes > MAX_ATTEMPT_ID || highBytes < 0) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " high bytes " + highBytes + " exceed");
+ }
+
+ long id = (highBytes << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + lowBytes;
+ LOG.info("convertTaskAttemptIdToLong id is {}", id);
+ LOG.info("lowBytes id is {}", lowBytes);
+ LOG.info("highBytes id is {}", highBytes);
+ return id;
+ }
+
+ public static long convertTaskAttemptIdToLong(int taskId, int taskAttemptID, int appAttemptId) {
+ long lowBytes = taskId;
+ if (lowBytes > Constants.MAX_TASK_ATTEMPT_ID) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " low bytes " + lowBytes + " exceed");
+ }
+ if (appAttemptId < 1) {
+ throw new RssException("appAttemptId " + appAttemptId + " is wrong");
+ }
+ long highBytes = (long)taskAttemptID - (appAttemptId - 1) * 1000;
+ if (highBytes > MAX_ATTEMPT_ID || highBytes < 0) {
+ throw new RssException("TaskAttempt " + taskAttemptID + " high bytes " + highBytes + " exceed");
+ }
+ return (highBytes << (Constants.TASK_ATTEMPT_ID_MAX_LENGTH + Constants.PARTITION_ID_MAX_LENGTH)) + lowBytes;
+ }
+
+ public static Roaring64NavigableMap fetchAllRssTaskIds(Set<InputAttemptIdentifier> successMapTaskAttempts,
Review Comment:
ok, we corrected it
--
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
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#issuecomment-1551190235
## [Codecov](https://app.codecov.io/gh/apache/incubator-uniffle/pull/890?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#890](https://app.codecov.io/gh/apache/incubator-uniffle/pull/890?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (768c5dd) into [master](https://app.codecov.io/gh/apache/incubator-uniffle/commit/ec9c278fc91f88247db87d848b0106f10132be02?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (ec9c278) will **decrease** coverage by `0.09%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #890 +/- ##
============================================
- Coverage 56.58% 56.50% -0.09%
+ Complexity 2175 2040 -135
============================================
Files 327 293 -34
Lines 15981 12935 -3046
Branches 1263 1200 -63
============================================
- Hits 9043 7309 -1734
+ Misses 6430 5229 -1201
+ Partials 508 397 -111
```
[see 65 files with indirect coverage changes](https://app.codecov.io/gh/apache/incubator-uniffle/pull/890/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197780407
##########
client-tez/src/main/java/org/apache/tez/common/TezClassLoader.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.common;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * ClassLoader to allow addition of new paths to classpath in the runtime.
+ *
+ * It uses URLClassLoader with this class' classloader as parent classloader.
+ * And hence first delegates the resource loading to parent and then to the URLs
+ * added. The process must be setup to use by invoking setupTezClassLoader() which sets
+ * the global TezClassLoader as current thread context class loader. All threads
+ * created will inherit the classloader and hence will resolve the class/resource
+ * from TezClassLoader.
+ */
+
+public class TezClassLoader extends URLClassLoader {
Review Comment:
Install the tez class loader, which can be used add new resources
click below to refer more:
https://github.com/apache/tez/blob/ccad71dabceb850b58ff38e77fd206e68f62ce4b/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java#L2338
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197794651
##########
client-tez/pom.xml:
##########
@@ -1,86 +1,91 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>uniffle-parent</artifactId>
<groupId>org.apache.uniffle</groupId>
- <version>0.8.0-SNAPSHOT</version>
+ <version>0.7.0</version>
Review Comment:
Ok, we corrected it.
##########
pom.xml:
##########
@@ -1625,10 +1625,13 @@
</dependencyManagement>
</profile>
<profile>
- <id>tez</id>
+ <id>tez0.9</id>
Review Comment:
Ok, we corrected it.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197803486
##########
client-tez/src/main/java/org/apache/tez/common/GetShuffleServerResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Objects;
+
+import org.apache.hadoop.io.Writable;
+
+public class GetShuffleServerResponse implements Writable {
Review Comment:
But we use this class in project multi times
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1197989893
##########
client-tez/src/test/java/org/apache/tez/common/TezIdHelperTest.java:
##########
@@ -0,0 +1,16 @@
+package org.apache.tez.common;
Review Comment:
Could you add the Apache license?
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Tez Shuffle Read supporting Remote Shuffle related common and utils
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198476453
##########
client-tez/src/main/java/org/apache/tez/common/TezClassLoader.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.common;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * ClassLoader to allow addition of new paths to classpath in the runtime.
+ *
+ * It uses URLClassLoader with this class' classloader as parent classloader.
+ * And hence first delegates the resource loading to parent and then to the URLs
+ * added. The process must be setup to use by invoking setupTezClassLoader() which sets
+ * the global TezClassLoader as current thread context class loader. All threads
+ * created will inherit the classloader and hence will resolve the class/resource
+ * from TezClassLoader.
+ */
+
+public class TezClassLoader extends URLClassLoader {
Review Comment:
We removed the file in this commit, battle it in next MR/PR
--
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
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198511763
##########
client-tez/pom.xml:
##########
@@ -34,53 +35,74 @@
<dependencies>
<dependency>
- <groupId>org.apache.uniffle</groupId>
- <artifactId>rss-client</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ <version>${tez.version}</version>
</dependency>
-
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-library</artifactId>
+ <version>${tez.version}</version>
</dependency>
<dependency>
- <groupId>net.jpountz.lz4</groupId>
- <artifactId>lz4</artifactId>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-runtime-internals</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ <version>${tez.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-api</artifactId>
+ <version>${tez.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
-
<dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
</dependency>
<dependency>
- <groupId>io.grpc</groupId>
- <artifactId>grpc-netty-shaded</artifactId>
- <version>${grpc.version}</version>
+ <groupId>org.apache.uniffle</groupId>
+ <artifactId>rss-client</artifactId>
</dependency>
<dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-all</artifactId>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
- <groupId>io.grpc</groupId>
- <artifactId>grpc-protobuf</artifactId>
+ <groupId>net.jpountz.lz4</groupId>
+ <artifactId>lz4</artifactId>
</dependency>
+
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>${protobuf.version}</version>
</dependency>
+
<dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>com.github.luben</groupId>
- <artifactId>zstd-jni</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${hadoop.version}</version>
Review Comment:
ditto.
--
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
[GitHub] [incubator-uniffle] lifeSo commented on a diff in pull request #890: [#872] feat(tez): Add the common and utils class
Posted by "lifeSo (via GitHub)" <gi...@apache.org>.
lifeSo commented on code in PR #890:
URL: https://github.com/apache/incubator-uniffle/pull/890#discussion_r1198656160
##########
client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java:
##########
@@ -0,0 +1,451 @@
+/*
+ * 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.common;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+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.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleWriteClient;
+import org.apache.uniffle.client.factory.ShuffleClientFactory;
+import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.Constants;
+
+public class RssTezUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RssTezUtils.class);
+
+ private static final int MAX_ATTEMPT_LENGTH = 6;
+ private static final long MAX_ATTEMPT_ID = (1 << MAX_ATTEMPT_LENGTH) - 1;
+
+ public static final String HOST_NAME = "hostname";
+
+ public static final String PLUS_DELIMITER = "+";
+ public static final String UNDERLINE_DELIMITER = "_";
+ public static final String COLON_DELIMITER = ":";
+ public static final String COMMA_DELIMITER = ",";
+
+ // constant to compute shuffle id
+ private static final int VERTEX_ID_MAPPING_MAX_ID = 500;
+ private static final String VERTEX_ID_MAPPING_MAP = "Map";
+ private static final String VERTEX_ID_MAPPING_REDUCER = "Reducer";
+ private static final int VERTEX_ID_MAPPING_MAGIC = 600;
+ private static final int SHUFFLE_ID_MAGIC = 1000;
+
+
+
+ private RssTezUtils() {
+ }
+
+ public static ShuffleWriteClient createShuffleClient(Configuration conf) {
+ int heartBeatThreadNum = conf.getInt(RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM,
+ RssTezConfig.RSS_CLIENT_HEARTBEAT_THREAD_NUM_DEFAULT_VALUE);
+ int retryMax = conf.getInt(RssTezConfig.RSS_CLIENT_RETRY_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_MAX_DEFAULT_VALUE);
+ long retryIntervalMax = conf.getLong(RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX,
+ RssTezConfig.RSS_CLIENT_RETRY_INTERVAL_MAX_DEFAULT_VALUE);
+ String clientType = conf.get(RssTezConfig.RSS_CLIENT_TYPE,
+ RssTezConfig.RSS_CLIENT_TYPE_DEFAULT_VALUE);
+ int replicaWrite = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_WRITE,
+ RssTezConfig.RSS_DATA_REPLICA_WRITE_DEFAULT_VALUE);
+ int replicaRead = conf.getInt(RssTezConfig.RSS_DATA_REPLICA_READ,
+ RssTezConfig.RSS_DATA_REPLICA_READ_DEFAULT_VALUE);
+ int replica = conf.getInt(RssTezConfig.RSS_DATA_REPLICA,
+ RssTezConfig.RSS_DATA_REPLICA_DEFAULT_VALUE);
+ boolean replicaSkipEnabled = conf.getBoolean(RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED,
+ RssTezConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE);
+ int dataTransferPoolSize = conf.getInt(RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE,
+ RssTezConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
+ int dataCommitPoolSize = conf.getInt(RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE,
+ RssTezConfig.RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE);
+ ShuffleWriteClient client = ShuffleClientFactory
+ .getInstance()
+ .createShuffleWriteClient(clientType, retryMax, retryIntervalMax,
+ heartBeatThreadNum, replica, replicaWrite, replicaRead, replicaSkipEnabled,
+ dataTransferPoolSize, dataCommitPoolSize);
+ return client;
+ }
+
+ public static long getInitialMemoryRequirement(Configuration conf, long maxAvailableTaskMemory) {
+ long initialMemRequestMb = conf.getLong(RssTezConfig.RSS_RUNTIME_IO_SORT_MB,
+ RssTezConfig.RSS_DEFAULT_RUNTIME_IO_SORT_MB);
+ LOG.info("InitialMemRequestMb is {}", initialMemRequestMb);
+ LOG.info("MaxAvailableTaskMemory is {}", maxAvailableTaskMemory);
+ long reqBytes = initialMemRequestMb << 20;
+ Preconditions.checkArgument(initialMemRequestMb > 0 && reqBytes < maxAvailableTaskMemory,
+ RssTezConfig.RSS_RUNTIME_IO_SORT_MB + initialMemRequestMb
+ + " should be " + "larger than 0 and should be less than the available task memory (MB):"
+ + (maxAvailableTaskMemory >> 20));
+ LOG.info("Requested BufferSize (" + TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB
+ + ") : " + initialMemRequestMb);
+ return reqBytes;
+ }
+
+ public static String uniformPartitionHostInfo(Map<Integer, List<ShuffleServerInfo>> map) {
+ List<String> res = new ArrayList<>();
+ String tmp;
+ Set<Integer> pidSet = map.keySet();
+ for (Integer pid : pidSet) {
+ for (ShuffleServerInfo shuffleServerInfo : map.get(pid)) {
+ tmp = pid + UNDERLINE_DELIMITER + shuffleServerInfo.getHost() + COLON_DELIMITER + shuffleServerInfo.getNettyPort();
+ res.add(tmp);
+ }
+ }
+ return org.apache.commons.lang.StringUtils.join(res, COMMA_DELIMITER);
+ }
+
+ public static Map<String, List<String>> uniformServerToPartitions(String partitionToServers) {
+ Map<String, List<String>> serverToPartitions = new HashMap<>();
+ List<String> list;
+
+ String[] pidWithWorkerInfos = partitionToServers.split(COMMA_DELIMITER);
+ for (String pidWithWorkerInfo : pidWithWorkerInfos) {
+ String[] pidUnderLineWorkerInfo = pidWithWorkerInfo.split(UNDERLINE_DELIMITER);
+ if (serverToPartitions.containsKey(pidUnderLineWorkerInfo[1])) {
+ list = serverToPartitions.get(pidUnderLineWorkerInfo[1]);
+ list.add(pidUnderLineWorkerInfo[0]);
+ } else {
+ list = new ArrayList<>();
+ list.add(pidUnderLineWorkerInfo[0]);
+ serverToPartitions.put(pidUnderLineWorkerInfo[1], list);
+ }
+ }
+
+ return serverToPartitions;
+ }
+
+ public static String uniformServerToPartitions(Map<String, List<String>> map) {
+ List<String> res = new ArrayList<>();
+ Set<String> keySet = map.keySet();
+ for (String s : keySet) {
+ String join = org.apache.commons.lang.StringUtils.join(map.get(s), UNDERLINE_DELIMITER);
+ res.add(s + PLUS_DELIMITER + join);
+ }
+
+ return org.apache.commons.lang.StringUtils.join(res,COMMA_DELIMITER);
+ }
+
+ public static ApplicationAttemptId getApplicationAttemptId() {
+ String containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.name());
+ ContainerId containerId = ContainerId.fromString(containerIdStr);
+ return containerId.getApplicationAttemptId();
+ }
+
+ public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
+ if (uniqueIdentifier == null) {
+ throw new RuntimeException("uniqueIdentifier should not be null");
+ }
+ String[] ids = uniqueIdentifier.split("_");
+ return StringUtils.join(ids, "_", 0, 7);
+ }
+
+ public static int getInt(Configuration rssJobConf, Configuration mrJobCOnf, String key, int defaultValue) {
Review Comment:
ok, we removed the getInt method()
--
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