You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by et...@apache.org on 2022/12/21 07:17:32 UTC
[incubator-celeborn] branch main updated: [CELEBORN-157] Change prefix of configurations to celeborn. (#1104)
This is an automated email from the ASF dual-hosted git repository.
ethanfeng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 5aa959a3 [CELEBORN-157] Change prefix of configurations to celeborn. (#1104)
5aa959a3 is described below
commit 5aa959a335f93a4065a2667cc440d56853d5b608
Author: Ethan Feng <et...@apache.org>
AuthorDate: Wed Dec 21 15:17:28 2022 +0800
[CELEBORN-157] Change prefix of configurations to celeborn. (#1104)
---
CONTRIBUTING.md | 5 +++++
.../apache/celeborn/common/util/ShutdownHookManager.java | 2 +-
.../scala/org/apache/celeborn/common/CelebornConf.scala | 15 +++++++++++----
.../celeborn/common/internal/config/ConfigEntry.scala | 7 ++++---
.../apache/celeborn/common/metrics/MetricsConfig.scala | 5 +++--
.../org/apache/celeborn/common/rpc/netty/Dispatcher.scala | 2 +-
.../apache/celeborn/common/rpc/netty/NettyRpcEnv.scala | 2 +-
docs/configuration/index.md | 8 ++++----
docs/configuration/worker.md | 1 +
.../celeborn/service/deploy/worker/FetchHandler.scala | 2 +-
10 files changed, 32 insertions(+), 17 deletions(-)
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index dd63eeab..5b459cc9 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -8,6 +8,11 @@ This project uses check-style plugins. Run some checks before you create a new p
/dev/reformat
```
+If you have changed configuration, run following command to refresh docs.
+```shell
+UPDATE=1 build/mvn clean test -pl common -am -Pspark-3.3 -Dtest=none -DwildcardSuites=org.apache.celeborn.ConfigurationSuite
+```
+
## How to Contribute
For collaboration, feel free to contact us on [Slack](https://join.slack.com/t/apachecelebor-kw08030/shared_invite/zt-1ju3hd5j8-4Z5keMdzpcVMspe4UJzF4Q).
To report a bug, you can just open a ticket on [Jira](https://issues.apache.org/jira/projects/CELEBORN/issues)
diff --git a/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java b/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java
index efc12459..12552c52 100644
--- a/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java
+++ b/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java
@@ -46,7 +46,7 @@ import org.apache.celeborn.common.CelebornConf;
*
* <p>Unless a hook was registered with a shutdown explicitly set through {@link
* #addShutdownHook(Runnable, int, long, TimeUnit)}, the shutdown time allocated to it is set by the
- * configuration option `rss.shutdown.timeout`.
+ * configuration option `celeborn.worker.graceful.shutdown.timeout`.
*
* <p>Note: code refer to Hadoop's ShutdownHookManager.
*/
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 0dfbc3b8..1192836a 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -57,7 +57,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
private def loadFromMap(props: Map[String, String], silent: Boolean): Unit =
settings.synchronized {
- // Load any rss.* system properties
+ // Load any celeborn.* or rss.* system properties
for ((key, value) <- props if key.startsWith("celeborn.") || key.startsWith("rss.")) {
set(key, value, silent)
}
@@ -793,9 +793,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
}
}
- def partitionSplitMinimumSize: Long = {
- getSizeAsBytes("rss.partition.split.minimum.size", "1m")
- }
+ def partitionSplitMinimumSize: Long = get(PARTITION_SPLIT_MIN)
def hdfsDir: String = {
get(HDFS_DIR).map {
@@ -2140,6 +2138,15 @@ object CelebornConf extends Logging {
.checkValues(Set(PartitionSplitMode.SOFT.name, PartitionSplitMode.HARD.name))
.createWithDefault(PartitionSplitMode.SOFT.name)
+ val PARTITION_SPLIT_MIN: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.partitionSplit.min")
+ .withAlternative("rss.partition.split.minimum.size")
+ .categories("worker")
+ .doc("Min size for a partition to split")
+ .version("0.2.0")
+ .longConf
+ .createWithDefaultString("1m")
+
val BATCH_HANDLE_CHANGE_PARTITION_ENABLED: ConfigEntry[Boolean] =
buildConf("celeborn.shuffle.batchHandleChangePartition.enabled")
.withAlternative("rss.change.partition.batch.enabled")
diff --git a/common/src/main/scala/org/apache/celeborn/common/internal/config/ConfigEntry.scala b/common/src/main/scala/org/apache/celeborn/common/internal/config/ConfigEntry.scala
index 98b2625e..84be8043 100644
--- a/common/src/main/scala/org/apache/celeborn/common/internal/config/ConfigEntry.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/internal/config/ConfigEntry.scala
@@ -49,9 +49,10 @@ The followings are best practices of naming configs for some common cases:
/**
* An entry contains all meta information for a configuration.
*
- * When applying variable substitution to config values, only references starting with "rss." are
- * considered in the default namespace. For known RSS configuration keys (i.e. those created using
- * `ConfigBuilder`), references will also consider the default value when it exists.
+ * When applying variable substitution to config values, only references starting with "celeborn."
+ * or "rss." are considered in the default namespace. For known RSS configuration
+ * keys (i.e. those created using`ConfigBuilder`), references will also consider the default
+ * value when it exists.
*
* Variable expansion is also applied to the default values of config entries that have a default
* value declared as a string.
diff --git a/common/src/main/scala/org/apache/celeborn/common/metrics/MetricsConfig.scala b/common/src/main/scala/org/apache/celeborn/common/metrics/MetricsConfig.scala
index 77ca8b24..ac07f7cb 100644
--- a/common/src/main/scala/org/apache/celeborn/common/metrics/MetricsConfig.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/metrics/MetricsConfig.scala
@@ -47,10 +47,11 @@ private class MetricsConfig(conf: CelebornConf) extends Logging {
// Add default properties in case there's no properties file
setDefaultProperties(properties)
- loadPropertiesFromFile(conf.getOption("rss.metrics.conf"))
+ val configKey = "celeborn.metrics.conf"
+ loadPropertiesFromFile(conf.getOption(configKey))
+ val prefix = s"${configKey}."
// Also look for the properties in provided rss configuration
- val prefix = "rss.metrics.conf."
conf.getAll.foreach {
case (k, v) if k.startsWith(prefix) =>
properties.setProperty(k.substring(prefix.length()), v)
diff --git a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala
index 99f78bee..d905be2a 100644
--- a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala
@@ -205,7 +205,7 @@ private[celeborn] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) e
val availableCores =
if (numUsableCores > 0) numUsableCores
else Math.max(16, Runtime.getRuntime.availableProcessors())
- val numThreads = nettyEnv.conf.getInt("rss.rpc.dispatcher.numThreads", availableCores)
+ val numThreads = nettyEnv.conf.getInt("celeborn.rpc.dispatcher.numThreads", availableCores)
val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, "dispatcher-event-loop")
logInfo(s"Dispatcher numThreads: $numThreads")
for (i <- 0 until numThreads) {
diff --git a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
index 9f023c39..1e1681a2 100644
--- a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
@@ -53,7 +53,7 @@ class NettyRpcEnv(
private[celeborn] val transportConf = Utils.fromCelebornConf(
conf.clone,
TransportModuleConstants.RPC_MODULE,
- conf.getInt("rss.rpc.io.threads", numUsableCores))
+ conf.getInt("celeborn.rpc.io.threads", numUsableCores))
private val dispatcher: Dispatcher = new Dispatcher(this, numUsableCores)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 9a5042dd..a93456f6 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -146,8 +146,8 @@ and restore the meta after restarting workers, also workers should keep a stable
`ShuffleClient` retry reading data. Users should set `celeborn.worker.graceful.shutdown.enabled` to `true` and
set below service port with stable port to support worker recover status.
```
-rss.worker.rpc.port
-rss.fetchserver.port
-rss.pushserver.port
-rss.replicateserver.port
+celeborn.worker.rpc.port
+celeborn.worker.fetch.port
+celeborn.worker.push.port
+celeborn.worker.replicate.port
```
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
index 44b62816..d9df2184 100644
--- a/docs/configuration/worker.md
+++ b/docs/configuration/worker.md
@@ -28,6 +28,7 @@ license: |
| celeborn.metrics.timer.slidingWindow.size | 4096 | The sliding window size of timer metric. | 0.2.0 |
| celeborn.shuffle.chuck.size | 8m | Max chunk size of reducer's merged shuffle data. For example, if a reducer's shuffle data is 128M and the data will need 16 fetch chunk requests to fetch. | 0.2.0 |
| celeborn.shuffle.minPartitionSizeToEstimate | 8mb | Ignore partition size smaller than this configuration of partition size for estimation. | 0.2.0 |
+| celeborn.shuffle.partitionSplit.min | 1m | Min size for a partition to split | 0.2.0 |
| celeborn.storage.hdfs.dir | <undefined> | HDFS dir configuration for Celeborn to access HDFS. | 0.2.0 |
| celeborn.test.pushdataTimeout | false | Wheter to test pushdata timeout | 0.2.0 |
| celeborn.worker.closeIdleConnections | false | Whether worker will close idle connections. | 0.2.0 |
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
index f5562d80..0fa0e51b 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
@@ -149,7 +149,7 @@ class FetchHandler(val conf: TransportConf) extends BaseMessageHandler with Logg
val chunksBeingTransferred = chunkStreamManager.chunksBeingTransferred
if (chunksBeingTransferred > conf.maxChunksBeingTransferred) {
val message = "Worker is too busy. The number of chunks being transferred " +
- s"$chunksBeingTransferred exceeds rss.shuffle.maxChunksBeingTransferred " +
+ s"$chunksBeingTransferred exceeds celeborn.shuffle.maxChunksBeingTransferred " +
s"${conf.maxChunksBeingTransferred}."
logError(message)
client.getChannel.writeAndFlush(