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 2023/05/08 09:07:45 UTC
[incubator-celeborn] branch main updated: [CELEBORN-570] Update docs about monitor and deployment. (#1478)
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 91b757555 [CELEBORN-570] Update docs about monitor and deployment. (#1478)
91b757555 is described below
commit 91b757555ea6632a10c02a7bdf3b383a74c8f3ea
Author: Ethan Feng <et...@apache.org>
AuthorDate: Mon May 8 17:07:42 2023 +0800
[CELEBORN-570] Update docs about monitor and deployment. (#1478)
---
README.md | 24 +++++++++---------
docs/cluster_planning.md | 37 ++++++++++++++++++++++++++++
docs/deploy.md | 64 +++++++++++++++++++++++++++++++++++++++---------
docs/monitoring.md | 23 ++++++++++++++++-
4 files changed, 124 insertions(+), 24 deletions(-)
diff --git a/README.md b/README.md
index ca8c3d362..c53a54a33 100644
--- a/README.md
+++ b/README.md
@@ -41,12 +41,12 @@ Celeborn Worker's slot count is decided by `total usable disk size / average shu
Celeborn worker's slot count decreases when a partition is allocated and increments when a partition is freed.
## Build
-1.Celeborn supports Spark 2.4/3.0/3.1/3.2/3.3 and flink 1.14.
+1.Celeborn supports Spark 2.4/3.0/3.1/3.2/3.3 and flink 1.14/1.15/1.17.
2.Celeborn tested under Java 8 environment.
Build Celeborn
```shell
-./build/make-distribution.sh -Pspark-2.4/-Pspark-3.0/-Pspark-3.1/-Pspark-3.2/-Pspark-3.3/-Pflink-1.14
+./build/make-distribution.sh -Pspark-2.4/-Pspark-3.0/-Pspark-3.1/-Pspark-3.2/-Pspark-3.3/-Pflink-1.14/-Pflink-1.15/-Pflink-1.17
```
package apache-celeborn-${project.version}-bin.tgz will be generated.
@@ -113,7 +113,8 @@ celeborn.master.port 9097
celeborn.metrics.enabled true
celeborn.worker.flush.buffer.size 256k
-celeborn.worker.storage.dirs /mnt/disk1:disktype=SSD,/mnt/disk2:disktype=HDD
+# Disk type is HDD by defaut.
+celeborn.worker.storage.dirs /mnt/disk1:disktype=SSD,/mnt/disk2:disktype=SSD
# If your hosts have disk raid or use lvm, set celeborn.worker.monitor.disk.enabled to false
celeborn.worker.monitor.disk.enabled false
```
@@ -121,10 +122,10 @@ celeborn.worker.monitor.disk.enabled false
EXAMPLE: HA cluster
```properties
# used by client and worker to connect to master
-celeborn.master.endpoints clb-1:9097,clb-2:9098,clb-3:9099
+celeborn.master.endpoints clb-1:9097,clb-2:9097,clb-3:9097
# used by master nodes to bootstrap, every node should know the topology of whole cluster, for each node,
-# `celeborn.ha.master.node.id` should be unique, and `celeborn.ha.master.node.<id>.host` is required
+# `celeborn.ha.master.node.id` should be unique, and `celeborn.ha.master.node.<id>.host` is required.
celeborn.ha.enabled true
celeborn.ha.master.node.id 1
celeborn.ha.master.node.1.host clb-1
@@ -141,7 +142,8 @@ celeborn.ha.master.ratis.raft.server.storage.dir /mnt/disk1/rss_ratis/
celeborn.metrics.enabled true
# If you want to use HDFS as shuffle storage, make sure that flush buffer size is at least 4MB or larger.
celeborn.worker.flush.buffer.size 256k
-celeborn.worker.storage.dirs /mnt/disk1:disktype=SSD,/mnt/disk2:disktype=HDD
+# Disk type is HDD by defaut.
+celeborn.worker.storage.dirs /mnt/disk1:disktype=SSD,/mnt/disk2:disktype=SSD
# If your hosts have disk raid or use lvm, set celeborn.worker.monitor.disk.enabled to false
celeborn.worker.monitor.disk.enabled false
```
@@ -150,9 +152,9 @@ Flink engine related configurations:
```properties
# if you are using Celeborn for flink, these settings will be needed
celeborn.worker.directMemoryRatioForReadBuffer 0.4
-celeborn.worker.directMemoryRatioToResume 0.5
+celeborn.worker.directMemoryRatioToResume 0.6
# these setting will affect performance.
-# If there is enough off-heap memory you can try to increase read buffers.
+# If there is enough off-heap memory, you can try to increase read buffers.
# Read buffer max memory usage for a data partition is `taskmanager.memory.segment-size * readBuffersMax`
celeborn.worker.partition.initial.readBuffersMin 512
celeborn.worker.partition.initial.readBuffersMax 1024
@@ -202,7 +204,7 @@ spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager
spark.serializer org.apache.spark.serializer.KryoSerializer
# celeborn master
-spark.celeborn.master.endpoints clb-1:9097,clb-2:9098,clb-3:9099
+spark.celeborn.master.endpoints clb-1:9097,clb-2:9097,clb-3:9097
spark.shuffle.service.enabled false
# options: hash, sort
@@ -227,10 +229,10 @@ spark.sql.adaptive.skewJoin.enabled true
Copy $CELEBORN_HOME/flink/*.jar to $FLINK_HOME/lib/
#### Flink Configuration
-TO use Celeborn, follow flink configurations should be added.
+TO use Celeborn, following flink configurations should be added.
```properties
shuffle-service-factory.class: org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory
-celeborn.master.endpoints: clb-1:9097,clb-2:9098,clb-3:9099
+celeborn.master.endpoints: clb-1:9097,clb-2:9097,clb-3:9097
celeborn.shuffle.batchHandleReleasePartition.enabled: true
celeborn.push.maxReqsInFlight: 128
diff --git a/docs/cluster_planning.md b/docs/cluster_planning.md
new file mode 100644
index 000000000..c7806b7e3
--- /dev/null
+++ b/docs/cluster_planning.md
@@ -0,0 +1,37 @@
+---
+license: |
+ 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.
+---
+
+# Cluster Planning
+
+## Node Spec
+
+Empirical size configs for Celeborn nodes
+The principle is to try to avoid any hardware(CPU, Memory, Disk Bandwidth/IOPS, Network
+Bandwidth/PPS)becoming the bottleneck.
+The goal is to let all the hardware usage be close to each other, for example let the disk
+IOPS/Bandwidth usage and network usage stay roughly the same so that data will be perfectly
+pipelined and no back-pressure will be triggered.
+
+The goal is hard to reach, and perhaps has a relationship with workload characteristics, and also
+Celeborn’s configs can have some impact. In our former experience, vCores: memory(GB): Bandwidth(
+Gbps): Disk IO (KIOps) is better to be 2: 5: 2: 1.
+We didn’t thoroughly conduct experiments on various configs(it’s hard to do so), so it’s merely a
+reference.
+
+## Worker Scale
+
+You need to estimate your cluster's max concurrent shuffle size(ES), and get the total usable disk
+space of a node(NS). The worker count can be `(ES * 2 / NS)`.
diff --git a/docs/deploy.md b/docs/deploy.md
index afa3b0e9d..8d21b26b9 100644
--- a/docs/deploy.md
+++ b/docs/deploy.md
@@ -14,7 +14,7 @@ license: |
limitations under the License.
---
-## Deploy Celeborn
+# Deploy Celeborn
1. Unzip the tarball to `$CELEBORN_HOME`
2. Modify environment variables in `$CELEBORN_HOME/conf/celeborn-env.sh`
@@ -39,7 +39,8 @@ celeborn.master.port 9097
celeborn.metrics.enabled true
celeborn.worker.flush.buffer.size 256k
-celeborn.worker.storage.dirs /mnt/disk1/,/mnt/disk2
+# Disk type is HDD by defaut.
+celeborn.worker.storage.dirs /mnt/disk1:disktype=SSD,/mnt/disk2:disktype=SSD
# If your hosts have disk raid or use lvm, set celeborn.worker.monitor.disk.enabled to false
celeborn.worker.monitor.disk.enabled false
```
@@ -47,31 +48,45 @@ celeborn.worker.monitor.disk.enabled false
EXAMPLE: HA cluster
```properties
# used by client and worker to connect to master
-celeborn.master.endpoints clb-1:9097,clb-2:9098,clb-3:9099
+celeborn.master.endpoints clb-1:9097,clb-2:9097,clb-3:9097
# used by master nodes to bootstrap, every node should know the topology of whole cluster, for each node,
-# `celeborn.ha.master.node.id` should be unique, and `celeborn.ha.master.node.<id>.host` is required
+# `celeborn.ha.master.node.id` should be unique, and `celeborn.ha.master.node.<id>.host` is required.
celeborn.ha.enabled true
celeborn.ha.master.node.id 1
celeborn.ha.master.node.1.host clb-1
celeborn.ha.master.node.1.port 9097
celeborn.ha.master.node.1.ratis.port 9872
celeborn.ha.master.node.2.host clb-2
-celeborn.ha.master.node.2.port 9098
-celeborn.ha.master.node.2.ratis.port 9873
+celeborn.ha.master.node.2.port 9097
+celeborn.ha.master.node.2.ratis.port 9872
celeborn.ha.master.node.3.host clb-3
-celeborn.ha.master.node.3.port 9099
-celeborn.ha.master.node.3.ratis.port 9874
+celeborn.ha.master.node.3.port 9097
+celeborn.ha.master.node.3.ratis.port 9872
celeborn.ha.master.ratis.raft.server.storage.dir /mnt/disk1/rss_ratis/
celeborn.metrics.enabled true
# If you want to use HDFS as shuffle storage, make sure that flush buffer size is at least 4MB or larger.
celeborn.worker.flush.buffer.size 256k
-celeborn.worker.storage.dirs /mnt/disk1/,/mnt/disk2
+# Disk type is HDD by defaut.
+celeborn.worker.storage.dirs /mnt/disk1:disktype=SSD,/mnt/disk2:disktype=SSD
# If your hosts have disk raid or use lvm, set celeborn.worker.monitor.disk.enabled to false
celeborn.worker.monitor.disk.enabled false
```
+Flink engine related configurations:
+```properties
+# if you are using Celeborn for flink, these settings will be needed
+celeborn.worker.directMemoryRatioForReadBuffer 0.4
+celeborn.worker.directMemoryRatioToResume 0.5
+# these setting will affect performance.
+# If there is enough off-heap memory you can try to increase read buffers.
+# Read buffer max memory usage for a data partition is `taskmanager.memory.segment-size * readBuffersMax`
+celeborn.worker.partition.initial.readBuffersMin 512
+celeborn.worker.partition.initial.readBuffersMax 1024
+celeborn.worker.readBuffer.allocationWait 10ms
+```
+
4. Copy Celeborn and configurations to all nodes
5. Start all services. If you install Celeborn distribution in same path on every node and your
cluster can perform SSH login then you can fill `$CELEBORN_HOME/conf/hosts` and
@@ -82,7 +97,7 @@ celeborn.worker.monitor.disk.enabled false
Start Celeborn worker
`$CELEBORN_HOME/sbin/start-worker.sh`
6. If Celeborn start success, the output of Master's log should be like this:
-```angular2html
+```
22/10/08 19:29:11,805 INFO [main] Dispatcher: Dispatcher numThreads: 64
22/10/08 19:29:11,875 INFO [main] TransportClientFactory: mode NIO threads 64
22/10/08 19:29:12,057 INFO [main] Utils: Successfully started service 'MasterSys' on port 9097.
@@ -104,7 +119,7 @@ WorkerRef: null
## Deploy Spark client
Copy $CELEBORN_HOME/spark/*.jar to $SPARK_HOME/jars/
-## Spark Configuration
+### Spark Configuration
To use Celeborn, following spark configurations should be added.
```properties
spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager
@@ -112,7 +127,7 @@ spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager
spark.serializer org.apache.spark.serializer.KryoSerializer
# celeborn master
-spark.celeborn.master.endpoints clb-1:9097,clb-2:9098,clb-3:9099
+spark.celeborn.master.endpoints clb-1:9097,clb-2:9097,clb-3:9097
spark.shuffle.service.enabled false
# options: hash, sort
@@ -132,3 +147,28 @@ spark.sql.adaptive.localShuffleReader.enabled false
spark.sql.adaptive.enabled true
spark.sql.adaptive.skewJoin.enabled true
```
+
+## Deploy Flink client
+Copy $CELEBORN_HOME/flink/*.jar to $FLINK_HOME/lib/
+
+### Flink Configuration
+TO use Celeborn, following flink configurations should be added.
+```properties
+shuffle-service-factory.class: org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory
+celeborn.master.endpoints: clb-1:9097,clb-2:9097,clb-3:9097
+
+celeborn.shuffle.batchHandleReleasePartition.enabled: true
+celeborn.push.maxReqsInFlight: 128
+
+# network connections between peers
+celeborn.data.io.numConnectionsPerPeer: 16
+# threads number may vary according to your cluster but do not set to 1
+celeborn.data.io.threads: 32
+celeborn.shuffle.batchHandleCommitPartition.threads: 32
+celeborn.rpc.dispatcher.numThreads: 32
+
+# floating buffers may need to change `taskmanager.network.memory.fraction` and `taskmanager.network.memory.max`
+taskmanager.network.memory.floating-buffers-per-gate: 4096
+taskmanager.network.memory.buffers-per-channel: 0
+taskmanager.memory.task.off-heap.size: 512m
+```
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 7d6bb0285..c9ac04c20 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -96,7 +96,9 @@ These metrics are exposed by Celeborn master.
- RegisteredShuffleCount
- IsActiveMaster
- PartitionSize
+ - The size of estimated shuffle partition size.
- OfferSlotsTime
+ - The time for a master to process offers slots request by register shuffle.
- namespace=CPU
- JVMCPUTime
@@ -119,12 +121,18 @@ These metrics are exposed by Celeborn worker.
- namespace=worker
- CommitFilesTime
+ - The time for a worker to flush buffers and close files related to specified shuffle.
- ReserveSlotsTime
- FlushDataTime
+ - The time for a worker to write a buffer which is 256KB by default to storage.
- OpenStreamTime
+ - The time for a worker to process openStream RPC and return StreamHandle.
- FetchChunkTime
+ - The time for a worker to fetch a chunk which is 8MB by default from a reduced partition.
- MasterPushDataTime
+ - The time for a worker to handle a pushData RPC sent from a celeborn client.
- SlavePushDataTime
+ - The time for a worker to handle a pushData RPC sent from a celeborn worker by replicating.
- WriteDataFailCount
- ReplicateDataFailCount
- ReplicateDataWriteFailCount
@@ -141,20 +149,33 @@ These metrics are exposed by Celeborn worker.
- MasterRegionFinishTime
- SlaveRegionFinishTime
- TakeBufferTime
- - TakeBufferTimeIndex
+ - The time for a worker to take out a buffer from a disk flusher.
- RegisteredShuffleCount
- SlotsAllocated
- NettyMemory
+ - The total amount of off-heap memory used by celeborn worker.
- SortTime
+ - The time for a worker to sort a shuffle file.
- SortMemory
+ - The memory used by sorting shuffle files.
- SortingFiles
- SortedFiles
- SortedFileSize
- DiskBuffer
+ - The memory occupied by pushData and pushMergedData which should be written to disk.
- PausePushData
+ - The count for a worker to stop receiving pushData from clients because of back pressure.
- PausePushDataAndReplicate
+ - The count for a worker to stop receiving pushData from clients and other workers because of back pressure.
- BufferStreamReadBuffer
+ - The memory used by credit stream read buffer.
- ReadBufferDispatcherRequestsLength
+ - The queue size of read buffer allocation requests.
+ - ReadBufferAllocatedCount
+ - Allocated read buffer count.
+ - CreditStreamCount
+ - Stream count for map partition reading streams.
+ - ActiveMapPartitionCount
- DeviceOSFreeCapacity(B)
- DeviceOSTotalCapacity(B)
- DeviceCelebornFreeCapacity(B)