Excessive-performance Distant Shuffle Service on Amazon EMR with Apache Celeborn

Organizations operating large-scale Apache Spark workloads usually face a trade-off between reaching decrease price and job reliability. These tradeoffs are extra distinguished when utilizing Amazon Elastic Compute Cloud (Amazon EC2) Spot Situations or when their jobs course of extremely skewed datasets. Three shuffle-related challenges drive the ache:

1. Spot interruptions set off expensive recomputation: Spot situations can cut back compute spend by as much as 90 % in comparison with On-Demand situations, however they are often reclaimed with solely two minutes of discover. When a Spark executor on a Spot Occasion is interrupted, its native shuffle knowledge is misplaced, and Spark should recompute whole upstream levels to regenerate that knowledge. For shuffle-heavy jobs processing terabytes of information, frequent interruptions trigger cascading recomputation and runtime delays, rapidly eroding the financial savings that made Spot engaging within the first place.
2. Native shuffle storage causes cluster-wide over-provisioning: In YARN-based Hadoop architectures, together with Amazon EMR on EC2, the Exterior Shuffle Service (ESS) shops shuffle knowledge regionally on every Node Supervisor’s node alongside the Spark executor that produced it. Each node should carry massive reminiscence and disk allocations to accommodate shuffle output, but only some EC2 nodes carry out a lot of the shuffle work. The remainder sit outsized and underused. It is a traditional coupled storage-compute drawback. By decoupling shuffle storage to a devoted, storage-optimized tier, you’ll be able to right-size your compute for precise calls for.
3. Shuffle knowledge safety leaves compute idle: To protect towards native shuffle knowledge loss, Spark’s scaling logic prevents nodes that also maintain shuffle knowledge from cutting down. EC2 nodes sit idle lengthy after Spark duties full. Information skew amplifies this impact: tail-end duties run far longer than typical ones, delaying shuffle reads and suspending scale-down throughout the cluster.

Collectively, these challenges pressure a tough alternative: cheaper infrastructure or predictable jobs. On this submit, we present how Apache Celeborn resolves this trade-off for Amazon EMR on EKS and Amazon EMR on EC2, bettering job reliability whereas unlocking further price financial savings.

What’s Apache Celeborn?

Apache Celeborn is an open-source Distant Shuffle Service (RSS) that solves the previous issues by decoupling shuffle knowledge from the executor lifecycle totally. It makes use of a Chief-Employee-Shopper structure: Chief nodes handle metadata, Employees learn and write shuffle blocks, and Purchasers combine with compute engines. As an alternative of writing shuffle output to native disks, Spark executors push knowledge to a shared, storage-optimized Celeborn cluster that persists shuffle knowledge independently of executor location. This implies EMR executor nodes can run on 100% Spot Situations. Spot reclamations now not trigger shuffle knowledge loss. Executors scale out and in freely with out triggering upstream recomputation.

Celeborn additionally supplies Raft-based excessive availability, per-job knowledge replication, and a pluggable shuffle supervisor that replaces Spark’s default mechanism with minimal configuration adjustments. With its push-based mannequin, Spark executors ship shuffle knowledge on to Celeborn staff, which cache and consolidate partitions. This reduces the N×M community connections in the course of the learn part, bettering each efficiency and stability at scale.

Picture 1: Push-based Distant Shuffle Service for Spark on EMR

Overview of resolution

On this submit, we present you deploy a Celeborn cluster alongside EMR on EKS and EMR on EC2. The answer additionally contains an observability stack to offer operational visibility into the Celeborn cluster. Metrics are collected by the AWS Distro for OpenTelemetry (ADOT) collector and routed to 2 monitoring paths. The AWS managed possibility makes use of Amazon Managed Service for Prometheus and Amazon Managed Grafana. The open supply possibility makes use of self-managed Prometheus with a built-in Grafana.

Apache Celeborn will be deployed in a number of methods relying in your operational necessities and scale. These two deployment patterns are the primary ones:

• Co-located on the identical cluster: Celeborn runs on the identical compute surroundings as Spark. That is probably the most simple operational mannequin, with no cross-cluster networking. The primary constraint is shared cluster lifecycle: any improve or termination impacts Celeborn and operating Spark jobs concurrently.
• Separate Celeborn cluster: Celeborn runs by itself EC2 or EKS cluster, totally remoted from Spark compute. That is probably the most operationally versatile mannequin and is the main target of this submit.

On this resolution, Celeborn runs on a devoted Amazon Elastic Kubernetes Service (Amazon EKS) cluster, separate from the EMR Spark environments. The 2 workloads have completely different useful resource profiles. Celeborn is storage and community I/O intensive, whereas Spark is CPU and reminiscence intensive. By separating them, every cluster can use occasion sorts optimized for its workload. It additionally improves impartial lifecycle administration, so you’ll be able to improve or scale Celeborn clusters with out disrupting Spark jobs, and the opposite means round.

Picture 2: Answer Structure

Because the structure diagram exhibits, two varieties of EMR deployment fashions, EMR on EKS and EMR on EC2, connect with a shared Celeborn cluster by means of an inside Community Load Balancer (NLB). Behind the scenes, Spark executors register their shuffle partitions to Celeborn staff by means of this connection, whereas reducers learn consolidated knowledge again in the course of the fetch part.

The next are the important thing design concerns for this resolution:

• Streamlined operation by a shared RSS mannequin: Celeborn runs on a devoted EKS cluster. This supplies lifecycle independence, permits every cluster to make use of workload-optimized occasion sorts, and permits a single Celeborn cluster to serve a number of EMR clusters as a shared service.
• Cross-cluster connectivity: Clusters reside in the identical Amazon Digital Non-public Cloud (Amazon VPC) and share non-public subnets. The AWS Load Balancer Controller on the Celeborn cluster provisions an inside NLB exposing its lively main pods on ports 9097 (RPC) and 9098 (dashboard). The NLB DNS title is VPC-resolvable, so any EMR cluster in the identical VPC can attain Celeborn by setting spark.celeborn.grasp.endpoints to the NLB handle.
• Restricted and secured networking: The Celeborn cluster solely permits inbound visitors from the EMR on EC2 and EMR on EKS clusters, sending shuffle knowledge and metrics over the community to the Celeborn cluster.
• State persistence: Main nodes preserve Celeborn’s coordination state by means of Raft consensus, which requires storage that survives pod restarts. Deploying them as StatefulSets with EBS-backed persistent quantity claims (PVC) lets a restarted main pod recuperate its Raft log and identification from sturdy storage moderately than ranging from scratch. Employees preserve shuffle knowledge on native NVMe occasion retailer for efficiency, however that is ephemeral. To guard knowledge loss on staff, every shuffle partition is replicated to 2 different staff by setting spark.celeborn.consumer.push.replicate.enabled=true.
• Observability: ADOT collector is deployed on the Celeborn EKS cluster. It scrapes Prometheus metrics from Celeborn’s pods, and concurrently distant writes them to 2 monitoring backends. Possibility 1 makes use of Amazon Managed Service for Prometheus because the metrics retailer, with Amazon Managed Grafana surfacing pre-built dashboards for Celeborn cluster well being and Java Digital Machine (JVM) metrics. This selection requires AWS IAM Identification Middle. Possibility 2 deploys a self-managed Prometheus stack with a built-in Grafana on the Celeborn EKS cluster, with Prometheus configured as a remote-write receiver for a similar ADOT collector. This selection fits environments with out IAM Identification Middle or these preferring a single open-source tooling.

Important configurations

The next two tables checklist the important thing configurations that make up the answer.

• Spark configuration (RSS consumer): Tells the Spark consumer to make use of Celeborn because the shuffle supervisor rather than Spark’s built-in implementation.
• Celeborn configuration (RSS server): Controls how the first and employee Celeborn pods function on Kubernetes.

Notice: The values within the following tables are reference defaults used on this walkthrough. Alter them based mostly in your workload necessities and cluster sizing.

The next desk highlights some key Spark configurations required to make use of Celeborn because the shuffle supervisor. You’ll be able to refer to those configurations utilized for every Spark submission technique within the scripts under:

The next desk supplies the server-side settings that management how the Celeborn main and employee pods function on Kubernetes. These values are set within the Helm values.yaml file.

Deploy the answer

This resolution comprises six layers, every of which depends on the earlier deployments. See the small print within the following deployment steps:

• Shared Infrastructure (Step 2).
• Celeborn Distant Shuffle Service set up (Step 3).
• Prepares pattern knowledge and creates EMR compute (Step 4, 5).
• Observability Layer (Step 6-7 and Step 9).
• Submit job (Step 8).
• Cleanup (Step 10).

Notice: This walkthrough creates billable AWS assets, together with Amazon EKS clusters, EC2 situations, Amazon Managed Grafana, Amazon Managed Service for Prometheus, and a Community Load Balancer. To keep away from ongoing prices, comply with the cleanup directions on the finish of this submit.

Stipulations

Earlier than you deploy this resolution, ensure that the next stipulations are in place:

• Entry to a sound AWS account with AWS Organizations and IAM Identification Middle enabled.
• The AWS Command Line Interface (AWS CLI) put in in your native machine.
• Git, kubectl, Helm 3 and curl, and Python 3 utilities put in in your native machine.
• Docker put in and operating.
• Permission to create AWS assets, resembling Amazon EKS clusters, AWS CloudFormation stacks, AWS Identification and Entry Administration (IAM) roles, VPC assets, and Amazon EMR clusters.
• Familiarity with Kubernetes, Apache Spark, Apache Celeborn, Amazon EKS, Amazon EMR on Amazon EKS, and Amazon EMR on Amazon EC2.

Deployment steps

Step 1: Clone from the supply repository

Clone the repository to your native machine and set the AWS_REGION:

git clone https://github.com/aws-samples/sample-emr-celeborn-shuffle-service.git
cd sample-emr-celeborn-shuffle-service
export AWS_REGION=

Step 2: Deploy the shared infrastructure

This step creates the core AWS assets, together with the VPC, AWS Key Administration Service (AWS KMS) key, safety teams, and Amazon Easy Storage Service (Amazon S3) bucket.

Step 3: Deploy the Celeborn cluster

This step provisions a devoted EKS cluster for Celeborn and exposes it by means of an inside NLB. It follows safety finest practices by protecting the EKS endpoint non-public, permitting public entry solely from a deployment workstation IP, and enabling encryption for secrets and techniques plus logging for all cluster parts.

Step 4: Put together the pattern knowledge

Execute the next script to generate pattern knowledge:

./spark-jobs/setup-data.sh

Step 5. Deploy a compute engine (select one or each)

Create at the least one of many following EMR deployment fashions to run Spark jobs.

Possibility A: EMR on EKS

Possibility B: EMR on EC2

Step 6. Deploy observability

To watch the Celeborn cluster, deploy one of many observability choices. After deployment, a Grafana URL and login particulars can be found within the .environment-info file positioned on the repository’s root listing. Comply with the directions in Step 9 to check in to your Grafana dashboard. You will notice two pre-built dashboards on the Grafana internet UI:

• Celeborn Cluster Overview: Lively shuffles, employee standing, disk utilization, and reminiscence utilization.
• Celeborn JVM Metrics: Heap utilization, rubbish assortment, and thread exercise.

Possibility A: AWS managed providers

This selection deploys an Amazon Managed Service for Prometheus workspace for metrics storage and an Amazon Managed Grafana workspace to host dashboards. Amazon Managed Grafana requires IAM Identification Middle, so first allow it on the group degree and create an SSO person:

export SSO_USER_EMAIL=
./observability/amp-amg/setup-sso.sh

Then deploy the stack:

./observability/amp-amg/deploy.sh

Possibility B: Open-source software

This selection deploys a Prometheus stack, together with a built-in Grafana service, onto the Celeborn EKS cluster within the monitoring namespace.

./observability/prometheus-grafana/deploy.sh

Step 7: Deploy ADOT Collector

The ADOT collector scrapes Prometheus metrics from Celeborn pods and remote-writes them to all lively monitoring backends: Amazon Managed Service for Prometheus, open-source Prometheus, or each.

./observability/adot/deploy.sh

Step 8: Submit a Spark job

The pattern job is a PySpark word-count utility that creates shuffle by means of groupBy and orderBy operations. Each EMR deployment choices under are configured with Celeborn because the shuffle supervisor.

Possibility A: Utilizing EMR on EKS

Submit the job utilizing the StartJobRun API:

./emr-on-eks/submit-emr-api.sh

This code snippet exhibits the core a part of the script (see the complete model):

aws emr-containers start-job-run
  --virtual-cluster-id "${VIRTUAL_CLUSTER_ID}"
  --name "${job_name}"
  --execution-role-arn "${JOB_EXECUTION_ROLE_ARN}"
  --release-label "${EMR_RELEASE_LABEL}"
  --job-driver "{
    "sparkSubmitJobDriver": {
      "entryPoint": "${input_path}",
      "entryPointArguments": ["${data_input}", "${data_output}"],
      "sparkSubmitParameters": "
        --conf spark.shuffle.supervisor= 
        org.apache.spark.shuffle.celeborn.SparkShuffleManager
        ...
      "
    }
  }"

Alternatively, submit the job utilizing a Spark Operator:

./emr-on-eks/submit-spark-operator.sh

The script routinely generates a SparkApplication manifest then applies it to EMR on EKS. For instance, kubectl apply -f your-job-manifest-name.yaml

Possibility B: Utilizing EMR on EC2

Submit the job as an EMR Step by means of the Steps API:

./emr-on-ec2/submit-job.sh

The next snippet exhibits the core API name used by the script:

aws emr add-steps
  --cluster-id "${CLUSTER_ID}"
  --region "${AWS_REGION}"
  --steps "Sort=Spark,
  Title=${JOB_NAME},
  ActionOnFailure=CONTINUE,
  Args=[
    --deploy-mode,client,
    --conf,spark.shuffle.manager= 
    org.apache.spark.shuffle.celeborn.SparkShuffleManager,
    ...
  ]"

Step 9: Evaluate the Grafana dashboard for distant shuffle metrics

The Grafana endpoint is dynamically generated at deploy time and is exclusive to your deployment. To entry it, open the .environment-info file on the repository root. This file comprises the Grafana URL together with login directions. Sign up utilizing the credentials listed there, then navigate to the Celeborn Cluster Overview dashboard to watch distant shuffle metrics in actual time. A pattern Grafana dashboard screenshot is proven under:

Picture 3: Grafana dashboard for Celeborn Metrics

Step 10. Cleansing up

To keep away from incurring future prices, run the cleanup script from the foundation listing:

This script detects which parts are deployed and tears them down in reverse dependency order, routinely skipping parts that aren’t current. Shared infrastructure is at all times deleted final, since different parts rely on it.

WARNING: This can completely take away all assets created beforehand, together with any knowledge saved in S3 buckets and configurations. The motion can’t be undone. Ensure you have backed up any knowledge you want to retain earlier than continuing.

Issues for manufacturing implementation

This submit demonstrates a working end-to-end structure for integrating Celeborn with Amazon EMR. Earlier than taking this sample to manufacturing, contemplate the next areas.

1. Safety

The next concerns provide help to safe a Celeborn deployment throughout knowledge isolation, encryption, and entry management.

1.1. Shuffle knowledge isolation between groups

Celeborn partitions shuffle knowledge by utility ID, which is designed to forestall jobs from by chance studying one another’s knowledge. That is enough when all jobs share the identical belief boundary. For multi-team deployments the place knowledge privateness is required, a devoted Celeborn cluster per group is the best isolation boundary: every group will get its personal NLB and safety group, and shuffle knowledge by no means co-mingles on the infrastructure degree.

1.2. Information in transit

In our implementation, shuffle knowledge travels over plain TCP between Spark executors and Celeborn staff. Entry is restricted to nodes utilizing safety teams. The inner NLB isn’t reachable exterior the VPC, and safety group guidelines block all different intra-VPC visitors.

For workloads requiring encryption in transit, Celeborn helps TLS on each RPC and knowledge channels by means of the celeborn.ssl.* configuration. It is a cluster-wide setting that applies to all jobs on the cluster and should be enabled server-side in celeborn-defaults.conf.

1.3. Information at relaxation

EBS volumes (used for Celeborn chief pod Raft state) are encrypted with the shared AWS KMS key. NVMe occasion retailer volumes on Celeborn employee nodes are ephemeral and never encrypted by default. Shuffle knowledge written to NVMe shouldn’t be protected by AWS KMS. For compliance necessities mandating encryption at relaxation, think about using EBS-backed employee storage as an alternative of occasion retailer, the place employee pods mount PersistentVolumeClaim (by means of volumeClaimTemplates) that reference the encrypted gp3 StorageClass. This comes at the price of decrease I/O throughput in comparison with native NVMe.

1.4. EKS secrets and techniques encryption

EKS clusters encrypt Kubernetes secrets and techniques at relaxation utilizing AWS KMS (EncryptionConfig within the cluster CloudFormation templates). This covers Kubernetes API objects however not application-level shuffle knowledge.

1.5. Software-level authorization

Our deployment enforces entry management within the community layer (that’s, VPC and safety group guidelines). Celeborn additionally supplies an application-level authorization framework, which is disabled by default. Turning it on provides a second degree of safety management the place solely purposes presenting legitimate credentials can register with the cluster. That is really helpful for manufacturing deployments the place a number of workloads or groups share the identical VPC subnets, making certain that community proximity alone doesn’t grant shuffle service entry.

Configuration for the Celeborn server:

celeborn.auth.enabled true # Allow SASL utility authentication
celeborn.inside.port.enabled true # Required when enabling SASL authentication

Configuration to allow authentication in each Spark app:

--conf spark.celeborn.auth.enabled=true

2. Autoscaling

You’ll be able to scale Celeborn staff or main pods utilizing kubectl. For instance:

kubectl scale statefulsets celeborn-worker -n celeborn --replicas=6
kubectl scale statefulsets celeborn-master -n celeborn --replicas=2

• Celeborn staff register with a main node once they begin, so scaling out is secure even whereas the cluster is operating. Scaling in, nevertheless, requires extra care. Eradicating a employee pod throughout an lively job might trigger shuffle knowledge loss except celeborn.consumer.push.replicate.enabled=true is enabled. To cut back the danger of unintended disruption throughout node scale-in or upgrades, add a Pod Disruption Funds to forestall a number of staff from being evicted on the identical time.
• Spark executors: Dynamic Useful resource Allocation (DRA) is enabled in pattern Spark jobs (spark.dynamicAllocation.enabled=true). This implies the variety of executors can scale routinely throughout the configured minExecutors and maxExecutors vary.

3. Resiliency

• Main Node HA: 3-replica Raft quorum with EBS-backed sturdy state is designed to tolerate one chief node failure with out job interruption.
• Employee replication: the setting celeborn.consumer.push.replicate.enabled=true copies every shuffle partition to 2 staff, designed to tolerate a single employee failure mid-job. With out replication, a employee failure causes a fetch failure and job retry.
• Pod Disruption Budgets (PDB): not configured on this demo. Add PDBs for Celeborn’s StatefulSets to forestall simultaneous eviction throughout node upgrades or scale-in.
• Multi-AZ placement: podAntiAffinity (most popular, weight 100) spreads pods throughout nodes and Availability Zones. For strict Availability Zone isolation, change to requiredDuringSchedulingIgnoredDuringExecution.

4. Monitoring and alerting

Take into account extending the Grafana dashboards with the alerting guidelines on:

• Lively shuffle partition rely: indicator of job load.
• Employee disk utilization: stop shuffle storage exhaustion.
• Push failure fee: early sign of connectivity or capability points.
• Celeborn main node failover: management adjustments point out Raft instability.

Conclusion

On this submit, we confirmed deploy Apache Celeborn as a Distant Shuffle Service on Amazon EKS and combine it with Amazon EMR on EKS and Amazon EMR on EC2. By decoupling shuffle storage from Spark’s compute, this structure delivers resilience to node failures, eliminates disk rivalry, and permits impartial scaling of storage and compute tiers.

Operating Celeborn on a devoted cluster offers you lifecycle independence from Spark, lets a number of EMR clusters share a single shuffle service, and supplies fault tolerance by means of push-based shuffling, Raft-based excessive availability, and per-job knowledge replication. It additionally supplies automated fallback to Spark’s built-in shuffle throughout upkeep home windows. Cease selecting between price and reliability. With Celeborn on Amazon EMR, you get each.

For extra data, see the Amazon EMR on EKS documentation and the Apache Celeborn documentation. To discover the complete implementation, go to the aws-samples GitHub repository. In case you have questions or suggestions, depart us a remark.

In regards to the authors