how to deploy a Spark cluster on eRDMA-enhanced instances - Elastic Compute Service

Build a Spark cluster with eRDMA-enhanced ECS instances and run Benchmark to measure load processing performance at ultra-low latency.

Background

Benchmark is a performance benchmarking tool that measures load execution time, transmission rate, throughput, and resource utilization.

Step 1: Prepare the environment

Set up the Hadoop cluster, install required drivers, and configure eRDMA before running the Benchmark test.

Prepare a Hadoop environment. Skip this step if a Hadoop cluster already exists.
- Hardware and software requirements
  Prepare the following Hadoop version, Spark version, and ECS instances:
  - Hadoop version: Hadoop 3.2.1.
  - Spark version: Spark 3.2.1.
  - ECS instances
    
    Instance type: See Overview.
    
    vCPUs per instance: 16.
    
    Instance count: four. One master node and three worker nodes.
- Installation
Log on to the master node. See Log on to a Linux instance using Workbench.
Configure eRDMA.
- Install the required drivers. See Enable eRDMA on enterprise-level instances.
- Configure network settings.
  1. Open the hosts file:
```
vim /etc/hosts
```
  2. Press I to enter Insert mode and modify the file content:
```
192.168.201.83 poc-t5m0        master1
192.168.201.84 poc-t5w0
192.168.201.86 poc-t5w1
192.168.201.85 poc-t5w2
```
    Note
    Replace the IP addresses with those of your actual eRDMA interfaces (ERIs).
  3. Press Esc, enter :wq, and press Enter to save and exit.
- Configure YARN settings.
  
  Note
  Skip this substep if the default NIC of the ECS instance supports eRDMA.
  1. Open the yarn-env.sh file:
```
cd /opt/hadoop-3.2.1/etc/hadoop
vim yarn-env.sh
```
  2. Press I to enter Insert mode and add the following content:
```
RDMA_IP=`ip addr show eth1 | grep "inet\b" | awk '{print $2}' | cut -d/ -f1`
export YARN_NODEMANAGER_OPTS="-Dyarn.nodemanager.hostname=$RDMA_IP"
```
    Note
    Replace eth1 with your actual ERI name.
  3. Press Esc, enter :wq, and press Enter to save and exit.
- Configure Spark.
  
  Note
  Skip this substep if the default NIC of the ECS instance supports eRDMA.
  1. Open the spark-env.sh file:
```
cd /opt/spark-3.2.1-bin-hadoop3.2/conf
vim spark-env.sh
```
  2. Press I to enter Insert mode and add the following content:
```
export SPARK_LOCAL_IP=`/sbin/ip addr show eth1 | grep "inet\b" | awk '{print $2}' | cut -d/ -f1`
```
    Note
    Replace eth1 with your actual ERI name.
  3. Press Esc, enter :wq, and press Enter to save and exit.
Start HDFS and YARN:
```
$HADOOP_HOME/sbin/start-all.sh
```

Step 2: Download the Benchmark installation package

Download and extract the Benchmark installation package to obtain the Spark eRDMA plug-in and dependencies.

Download the Benchmark installation package:

wget https://mracc-release.oss-cn-beijing.aliyuncs.com/erdma-spark/spark-erdma-jverbs.tar.gz

Decompress the spark-erdma-jverbs.tar.gz package:
```
tar -zxvf spark-erdma-jverbs.tar.gz
```
The package contains the following components:
- erdmalib: the native library required by the spark-erdma plug-in, corresponding to libdisni.so.
- plugin-sparkrdma: the Spark RDMA plug-in and dependency library, corresponding to spark-eRDMA-1.0-for-spark-3.2.1.jar and disni-2.1-jar-with-dependencies.jar.

Step 3: Run a Benchmark test

Configure IP routes and Spark settings, generate test data, and run the Benchmark test.

Modify IP routes.

Note
Skip this step if the default NIC of your ECS instance supports eRDMA.
```
route del -net 192.168.201.0 netmask 255.255.255.0 metric 0 dev eth0 && \
route add -net 192.168.201.0 netmask 255.255.255.0 metric 1000 dev eth0
```
Note
Replace the IP addresses with the gateway IP address of your actual ERI.

Configure Spark.

Open the spark-jverbs-erdma.conf file:

vim /opt/spark-3.2.1-bin-hadoop3.2/conf/spark-jverbs-erdma.conf

Press I to enter Insert mode and modify the file content:

spark.master yarn
spark.deploy-mode client
#driver
spark.driver.cores 4
spark.driver.memory 19g
#executor
spark.executor.instances 12
spark.executor.memory 10g
spark.executor.cores 4
spark.executor.heartbeatInterval   60s
#shuffle
spark.task.maxFailures 4
spark.default.parallelism 36
spark.sql.shuffle.partitions 192
spark.shuffle.compress            true
spark.shuffle.spill.compress      true

#other
spark.network.timeout 3600
spark.sql.broadcastTimeout 3600
spark.eventLog.enabled             false
spark.eventLog.dir                 hdfs://master1:9000/sparklogs
spark.eventLog.compress            true
spark.yarn.historyServer.address   master1:18080
spark.serializer                  org.apache.spark.serializer.KryoSerializer

#eRDMA
spark.driver.extraLibraryPath   /path/erdmalib
spark.executor.extraLibraryPath   /path/erdmalib
spark.driver.extraClassPath       /path/spark-eRDMA-1.0-for-spark-3.2.1.jar:/path/disni-2.1-jar-with-dependencies.jar
spark.executor.extraClassPath     /path/spark-eRDMA-1.0-for-spark-3.2.1.jar:/path/disni-2.1-jar-with-dependencies.jar
spark.shuffle.manager org.apache.spark.shuffle.sort.RdmaShuffleManager
spark.shuffle.sort.io.plugin.class org.apache.spark.shuffle.rdma.RdmaLocalDiskShuffleDataIO
spark.shuffle.rdma.recvQueueDepth  128

Note

Set spark.shuffle.compress to false for a better acceleration ratio.
The sample code uses Spark resource settings for an ECS instance with 32 vCPUs and 128 GB of memory. Adjust spark.executor.instances, spark.executor.memory, spark.executor.cores, and spark.sql.shuffle.partitions based on your cluster scale or instance specifications.

Press Esc, enter :wq, and press Enter to save and exit.

Generate data:

cd /opt/spark-3.2.1-bin-hadoop3.2/conf
spark-submit --properties-file /opt/spark-3.2.1-bin-hadoop3.2/conf/spark-normal.conf --class com.databricks.spark.sql.perf.tpcds.TPCDS_Bench_DataGen spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar hdfs://master1:9000/tmp/tpcds_400 tpcds_400 400 parquet

Note

400 specifies the data volume in GB. Adjust based on your cluster scale.

Run the Benchmark test:

spark-submit --properties-file /opt/spark-3.2.1-bin-hadoop3.2/conf/spark-jverbs-erdma.conf --class com.databricks.spark.sql.perf.tpcds.TPCDS_Bench_RunAllQuery spark-sql-perf_2.12-0.5.1-SNAPSHOT.jar all hdfs://master1:9000/tmp/tpcds_400 tpcds_400 /tmp/tpcds_400_result

The following output indicates the test is complete. The test result shows the load execution time of the Spark cluster.

22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 371.0 in stage 1.0 (TID 755) in 31 ms on c7re4x004 (executor 1) (372/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 372.0 in stage 1.0 (TID 756) in 29 ms on c7re4x004 (executor 1) (373/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 376.0 in stage 1.0 (TID 760) in 23 ms on c7re4x004 (executor 1) (374/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 374.0 in stage 1.0 (TID 758) in 27 ms on c7re4x004 (executor 1) (375/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 373.0 in stage 1.0 (TID 757) in 31 ms on c7re4x004 (executor 1) (376/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 377.0 in stage 1.0 (TID 761) in 24 ms on c7re4x004 (executor 1) (377/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 375.0 in stage 1.0 (TID 759) in 29 ms on c7re4x004 (executor 1) (378/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 381.0 in stage 1.0 (TID 765) in 53 ms on c7re4x004 (executor 1) (379/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 380.0 in stage 1.0 (TID 764) in 53 ms on c7re4x004 (executor 1) (380/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 379.0 in stage 1.0 (TID 763) in 56 ms on c7re4x004 (executor 1) (381/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 382.0 in stage 1.0 (TID 766) in 55 ms on c7re4x004 (executor 1) (382/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 378.0 in stage 1.0 (TID 762) in 56 ms on c7re4x004 (executor 1) (383/384)
22/06/10 15:35:40 INFO scheduler.TaskSetManager: Finished task 383.0 in stage 1.0 (TID 767) in 53 ms on c7re4x004 (executor 1) (384/384)
22/06/10 15:35:40 INFO cluster.YarnScheduler: Removed TaskSet 1.0, whose tasks have all completed, from pool
22/06/10 15:35:40 INFO scheduler.DAGScheduler: ResultStage 1 (runJob at SparkHadoopWriter.scala:78) finished in 1.599 s
22/06/10 15:35:40 INFO scheduler.DAGScheduler: Job 0 finished: runJob at SparkHadoopWriter.scala:78, took 7.308027 s
22/06/10 15:35:40 INFO io.SparkHadoopWriter: Job job_20220610153533_0002 committed.
22/06/10 15:35:40 INFO server.AbstractConnector: Stopped Spark@1cfd3b14{HTTP/1.1,[http/1.1]}{192.168.145.101:4040}
22/06/10 15:35:40 INFO ui.SparkUI: Stopped Spark web UI at http://c7re4x001:4040
22/06/10 15:35:41 INFO cluster.YarnClientSchedulerBackend: Interrupting monitor thread
22/06/10 15:35:41 INFO cluster.YarnClientSchedulerBackend: Shutting down all executors
22/06/10 15:35:41 INFO cluster.YarnSchedulerBackend$YarnDriverEndpoint: Asking each executor to shut down
22/06/10 15:35:41 INFO cluster.SchedulerExtensionServices: Stopping SchedulerExtensionServices
(serviceOption=None,
 services=List(),
 started=false)
22/06/10 15:35:41 INFO cluster.YarnClientSchedulerBackend: Stopped
22/06/10 15:35:41 INFO spark.MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
22/06/10 15:35:41 INFO memory.MemoryStore: MemoryStore cleared
22/06/10 15:35:41 INFO storage.BlockManager: BlockManager stopped
22/06/10 15:35:41 INFO storage.BlockManagerMaster: BlockManagerMaster stopped
22/06/10 15:35:41 INFO scheduler.OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
22/06/10 15:35:41 INFO spark.SparkContext: Successfully stopped SparkContext
22/06/10 15:35:41 INFO util.ShutdownHookManager: Shutdown hook called
22/06/10 15:35:41 INFO util.ShutdownHookManager: Deleting directory /mnt/data/spark_tmp/spark-d363e35b-badc-43ba-9fa9-476a9301f65c
22/06/10 15:35:41 INFO util.ShutdownHookManager: Deleting directory /tmp/spark-87ef4db1-7c2f-4f23-b989-f27b9bd9ac28

Note

To compare eRDMA performance, remove the spark-erdma plug-in configurations from the Spark conf directory or use a Spark cluster without eRDMA, run the same Benchmark test, and compare the results.