Amazon EMR Deployment UserGuide#
Amazon EMR (previously called Amazon Elastic MapReduce) is a managed cluster platform that simplifies running big data frameworks, such as Apache Hadoop and Apache Spark, on AWS to process and analyze vast amounts of data. The NVIDIA RAPIDS Accelerator for Apache Spark is available on EMR, allowing users to accelerate data processing and machine learning workloads using GPUs. This integration is fully supported by AWS and enables users to run their Spark workloads with optimized performance and efficiency.
Overview of Steps#
This guide provides step by step instructions for getting started with using the RAPIDS Accelerator for an AWS EMR cluster. We start with an Ubuntu based OS to run the AWS CLI for connecting and interacting with an EMR cluster and an s3 bucket. Once the cluster is created, we can then submit a GPU accelerated Spark job/application via Spark Submit. Instructions for using EMR studio with GPU accelerated instances are also provided.
Prerequisites#
Prior to getting started with the RAPIDS Accelerator for Apache Spark on AWS EMR ensure you have a the following prerequisites:
Ubuntu OS with internet access to AWS
NGC Account with NGC Catalog Access
AWS Cloud Tools installed and configured with an AWS key pair
The supported components for using the NVIDIA RAPIDS Accelerator with EMR are listed in the table below.
EMR |
Spark |
RAPIDS Accelerator jar |
cuDF jar |
xgboost4j-spark jar |
---|---|---|---|---|
6.10 |
3.3.1 |
rapids-4-spark_2.12-22.12.0-amzn-0.jar |
Bundled with rapids-4-spark |
xgboost4j-spark_3.0-1.4.2-0.3.0.jar |
For more details of supported applications, please see the EMR release notes. For more information on AWS EMR, please see the AWS documentation.
Connectivity#
A simple way to validate access to EMR is to list out the EMR clusters via
aws emr list-clusters
Create and Launch AWS EMR with GPU Nodes#
The following steps are based on the AWS EMR document Using the NVIDIA RAPIDS Accelerator for Spark. This document provides a basic configuration to get users started.
The my-configurations.json
installs the spark-rapids plugin on your cluster, configures YARN to use GPUs, configures Spark to use RAPIDS, and configures the YARN capacity scheduler. Create a local file called my-configurations.json with the following default configurations:
1[
2 {
3 "Classification":"spark",
4 "Properties":{
5 "enableSparkRapids":"true"
6 }
7 },
8 {
9 "Classification":"yarn-site",
10 "Properties":{
11 "yarn.nodemanager.resource-plugins":"yarn.io/gpu",
12 "yarn.resource-types":"yarn.io/gpu",
13 "yarn.nodemanager.resource-plugins.gpu.allowed-gpu-devices":"auto",
14 "yarn.nodemanager.resource-plugins.gpu.path-to-discovery-executables":"/usr/bin",
15 "yarn.nodemanager.linux-container-executor.cgroups.mount":"true",
16 "yarn.nodemanager.linux-container-executor.cgroups.mount-path":"/sys/fs/cgroup",
17 "yarn.nodemanager.linux-container-executor.cgroups.hierarchy":"yarn",
18 "yarn.nodemanager.container-executor.class":"org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor"
19 }
20 },
21 {
22 "Classification":"container-executor",
23 "Properties":{
24
25 },
26 "Configurations":[
27 {
28 "Classification":"gpu",
29 "Properties":{
30 "module.enabled":"true"
31 }
32 },
33 {
34 "Classification":"cgroups",
35 "Properties":{
36 "root":"/sys/fs/cgroup",
37 "yarn-hierarchy":"yarn"
38 }
39 }
40 ]
41 },
42 {
43 "Classification":"spark-defaults",
44 "Properties":{
45 "spark.plugins":"com.nvidia.spark.SQLPlugin",
46 "spark.sql.sources.useV1SourceList":"",
47 "spark.executor.resource.gpu.discoveryScript":"/usr/lib/spark/scripts/gpu/getGpusResources.sh",
48 "spark.executor.extraLibraryPath":"/usr/local/cuda/targets/x86_64-linux/lib:/usr/local/cuda/extras/CUPTI/lib64:/usr/local/cuda/compat/lib:/usr/local/cuda/lib:/usr/local/cuda/lib64:/usr/lib/hadoop/lib/native:/usr/lib/hadoop-lzo/lib/native:/docker/usr/lib/hadoop/lib/native:/docker/usr/lib/hadoop-lzo/lib/native",
49 "spark.executor.resource.gpu.amount":"1",
50 "spark.rapids.sql.concurrentGpuTasks":"2",
51 "spark.sql.files.maxPartitionBytes":"256m",
52 "spark.sql.adaptive.enabled":"true"
53 }
54 },
55 {
56 "Classification":"capacity-scheduler",
57 "Properties":{
58 "yarn.scheduler.capacity.resource-calculator":"org.apache.hadoop.yarn.util.resource.DominantResourceCalculator"
59 }
60 }
61]
Adaptive Query Execution (AQE) is enabled by default in Spark 3.2.0, however, the RAPIDS accelerator does not fully support AQE until EMR 6.10+. The setting can be toggled via spark.sql.adaptive.enabled
. For lower EMR versions, this setting should be set to false. More advanced configuration details can be found in the configuration documentation.
The my-bootstrap-action.sh
script opens cgroup permissions to YARN on your cluster for YARN to use GPUs. Create this file locally with the following content.
1#!/bin/bash
2set -ex
3sudo chmod a+rwx -R /sys/fs/cgroup/cpu,cpuacct
4sudo chmod a+rwx -R /sys/fs/cgroup/devices
Upload it into an S3 bucket and export the file path. Replace <my-bucket>
with your bucket name in the below command.
export BOOTSTRAP_ACTION_SCRIPT=s3://<my-bucket>/my-bootstrap-action.sh
This file will be used for Launching an EMR Cluster
Create a Key pair#
Prior to launching a cluster, a Key Pair is required at cluster creation to allow for SSH access to the cluster. If you have not already created a Key Pair, follow instructions on https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/create-key-pairs.html.
Update KEYNAME in the export statement below with your personal key pair (*.pem file). The key pair is needed to securely access the EMR cluster
Launch an EMR Cluster using AWS CLI#
You can use the AWS CLI to launch a cluster with one master node (m5.xlarge) and two g4dn.2xlarge worker nodes:
1export KEYNAME=my-key-pair
2export CLUSTER_NAME=spark-rapids-cluster
3export EMR_LABEL=emr-6.10.0
4export MASTER_INSTANCE_TYPE=m4.4xlarge
5export NUM_WORKERS=2
6export WORKER_INSTANCE_TYPE=g4dn.2xlarge
7export CONFIG_JSON_LOCATION=./my-configurations.json
Create the EMR cluster:
1aws emr create-cluster \
2--name $CLUSTER_NAME \
3--release-label $EMR_LABEL \
4--applications Name=Hadoop Name=Spark Name=Livy Name=JupyterEnterpriseGateway \
5--service-role EMR_DefaultRole \
6--ec2-attributes KeyName=$KEYNAME,InstanceProfile=EMR_EC2_DefaultRole \
7--instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=$MASTER_INSTANCE_TYPE \
8InstanceGroupType=CORE,InstanceCount=$NUM_WORKERS,InstanceType=$WORKER_INSTANCE_TYPE \
9--configurations file://$CONFIG_JSON_LOCATION \
10--bootstrap-actions Name='My Spark Rapids Bootstrap action',Path=$BOOTSTRAP_ACTION_SCRIPT
Output example:
1{
2 "ClusterId": "j-GEXQ0J54HTI0",
3 "ClusterArn": "arn:aws:elasticmapreduce:us-west-2:123456789:cluster/j-GEXQ0J54HTI0"
4}
Save the ClusterId for checking the cluster status in subsequent sections.
Create EMR Cluster with Specified Subnet (Optional)#
EMR Studio provides a Jupyter notebook environment that can be connected to EMR clusters. A Virtual Private Cloud (VPC) is required to secure the connection. This can be enforced by specifying the SubnetId during the EMR cluster creation step by including the SubnetId argument within the ec2-attributes option during aws emr create-cluster:
--ec2-attributes KeyName=$KEYNAME,InstanceProfile=EMR_EC2_DefaultRole,SubnetId=<gpu_enabled_subnet> \
To help determine which VPC SubnetId (subnet-*) to use, search for the VPC configuration page. From the left navigation menu on the VPC page, select “Virtual private cloud” >> “Your VPCs” >> Select existing or create new VPC. Use the SubnetId for us-west-2b or other Availability Zone where GPUs are available. The subnet Availability Zone can be determined from the Resource map:
Note
GPU Cluster creation time can take ~15+ min.
Validation#
Check Cluster status:
$ aws emr describe-cluster --cluster-id <cluster_id>
Below is an example output:
1{
2 "Cluster": {
3 "Id": "j-1I7SZJ8DRXCWY",
4 "Name": "sample-cluster",
5 "Status": {
6 "State": "WAITING",
7 "StateChangeReason": {
8 "Message": "Cluster ready to run steps."
9 },
10 "Timeline": {
11 "CreationDateTime": "2023-03-17T17:29:10.912000+00:00",
12 "ReadyDateTime": "2023-03-17T17:42:06.963000+00:00"
13 }
14 },
Connect to the EMR cluster master with the following command. Modify the command to include your <cluster_id>
and key pair.
aws emr ssh --cluster-id <cluster_id> --key-pair-file ~/.ssh/my-key-pair.pem
Running an example join operation using Spark Shell#
On the master node, activate the spark-shell by typing:
spark-shell
Run the following Scala code in Spark Shell:
1val data = 1 to 10000
2val df1 = sc.parallelize(data).toDF()
3val df2 = sc.parallelize(data).toDF()
4val out = df1.as("df1").join(df2.as("df2"), $"df1.value" === $"df2.value")
5out.count()
6out.explain()
Exit the spark-shell and the SSH session before continuing to the next step.
Spark Submit Jobs to a EMR Cluster Accelerated by GPUs#
EMR jobs can be initiated by logging into the master node and submitting applications via spark-submit
, just like with Apache Spark. By default, YARN operates in client mode. All applicable files referenced in spark-submit
should be accessible on the master node.
export SPARK_HOME=/usr/lib/spark
1$SPARK_HOME/bin/spark-submit \
2--class org.apache.spark.examples.SparkPi \
3--master yarn \
4$SPARK_HOME/examples/jars/spark-examples.jar \
51000
The console output will look like this:
The bottom section of the outputs are from exiting the application. The results of interest are:
123/03/16 20:44:04 INFO DAGScheduler: Job 0 finished: reduce at SparkPi.scala:38, took 34.622154 s
2Pi is roughly 3.1416558714165586
Spark History Server UI#
The Spark history server UI is on port 18080 and is accessible via port mapping. To view the spark logs, SSH into the master with port forwarding for port 18080 enabled.
The master node public IP can be found via the AWS web console UI. Select your EMR cluster, then click on the “Instances” tab as shown below. Select the link under the ID column that starts with “ig-”* for the Instance Group corresponding to the master node. The public IP address of the EC2 instance corresponding to the master node is shown in the “Public IP address” column (you may need to scroll to the right).
Use this IP address to update <master_node_public_ip>
in the command below:
ssh -i ~/.ssh/my-key-pair.pem hadoop@<master_node_public_ip> -L 18080:localhost:18080
Then open a browser window and navigate to: localhost:18080. Keep the SSH session open to maintain the port forwarding. The spark event logs can be downloaded via the web UI.
Cluster Cleanup#
The cluster can be terminated in the GUI by searching for EMR, then selecting your cluster and the Terminate button on the upper right.
Alternatively, the cluster can be terminated via the CLI with:
aws emr terminate-clusters --cluster-id <cluster_id>
If you started an EMR Studio session, the session can be stopped through the EMR Studio webpage. Select “Workspaces” from the left navigation panel, click on a Workspace that’s active, press the Actions button on the upper right, then select “Stop”.