利用 Apache Celeborn 减少 AWS EMR 集群中 Spot 实例回收导致的重算成本

AWSTemplateFormatVersion: '2010-09-09'
Description: Launch 3 EC2 instances with an existing key pair, security group, and an IAM role with permissions for S3 and DynamoDB operations. Execute a script from an S3 path during instance launch.

Parameters:
  KeyName:
    Description: Name of an existing EC2 key pair for SSH access
    Type: AWS::EC2::KeyPair::KeyName
  SecurityGroupId:
    Description: ID of an existing security group to associate with the instances
    Type: AWS::EC2::SecurityGroup::Id
  InstanceType:
    Description: EC2 instance type (M series)
    Type: String
    Default: m6g.large
    AllowedValues:
      - m6g.large
      - m6g.xlarge
      - m6g.2xlarge
  ScriptS3Path:
    Description: S3 path to the script to execute during instance launch
    Type: String

Resources:
  IAMRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - ec2.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess

  InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      Roles:
        - !Ref IAMRole

  EC2Instance1:
    Type: AWS::EC2::Instance
    Properties:
      ImageId: ami-063f1c6928a072c56  # Replace with your desired AMI
      InstanceType: !Ref InstanceType
      KeyName: !Ref KeyName
      SecurityGroupIds:
        - !Ref SecurityGroupId
      IamInstanceProfile: !Ref InstanceProfile
      UserData:
        Fn::Base64:
          !Sub |
            #!/bin/bash
            aws s3 cp ${ScriptS3Path} /tmp/script.sh
            chmod +x /tmp/script.sh
            /tmp/script.sh /tmp/output.log 2>&1

  EC2Instance2:
    Type: AWS::EC2::Instance
    Properties:
      ImageId: ami-063f1c6928a072c56 # Replace with your desired AMI
      InstanceType: !Ref InstanceType
      KeyName: !Ref KeyName
      SecurityGroupIds:
        - !Ref SecurityGroupId
      IamInstanceProfile: !Ref InstanceProfile
      UserData:
        Fn::Base64:
          !Sub |
            #!/bin/bash
            aws s3 cp ${ScriptS3Path} /tmp/script.sh
            chmod +x /tmp/script.sh
            /tmp/script.sh  /tmp/output.log 2>&1

  EC2Instance3:
    Type: AWS::EC2::Instance
    Properties:
      ImageId: ami-063f1c6928a072c56  # Replace with your desired AMI
      InstanceType: !Ref InstanceType
      KeyName: !Ref KeyName
      SecurityGroupIds:
        - !Ref SecurityGroupId
      IamInstanceProfile: !Ref InstanceProfile
      UserData:
        Fn::Base64:
          !Sub |
            #!/bin/bash
            aws s3 cp ${ScriptS3Path} /tmp/script.sh
            chmod +x /tmp/script.sh
            /tmp/script.sh >  /tmp/output.log 2>&1

Outputs:
  InstanceIds:
    Description: The IDs of the launched EC2 instances
    Value: !Join [',', [!Ref EC2Instance1, !Ref EC2Instance2, !Ref EC2Instance3]]

#!/bin/bash -xe
cd /home/ec2-user
sudo yum install java-1.8.0 -y

wget https://downloads.apache.org/celeborn/celeborn-0.4.1/apache-celeborn-0.4.1-bin.tgz

tar -xvf apache-celeborn-0.5.0-bin.tgz

cd /home/ec2-user/apache-celeborn-0.5.0-bin/conf

cp log4j2.xml.template log4j2.xml

cp celeborn-env.sh.template celeborn-env.sh

echo "celeborn.worker.storage.dirs=/home/ec2-user/apache-celeborn-0.5.0-bin/shuffle" > celeborn-defaults.conf


TABLE_EXISTS=$(aws dynamodb list-tables --query 'TableNames[?contains(@, `celeborn`)]' --region us-east-1 --output text)

if [ -n "$TABLE_EXISTS" ]; then
    echo "Table 'celeborn' exists."
    sleep 20
     IP_ADDRESSES=$(aws dynamodb scan \
    --table-name celeborn \
    --projection-expression "IPAddress" \
    --query "Items[*].IPAddress.S" \
    --region us-east-1 \
    --output text)
   sudo  /home/ec2-user/apache-celeborn-0.5.0-bin/sbin/start-worker.sh celeborn://$IP_ADDRESSES:9097
else
    echo "Table 'celeborn' does not exist. Creating table..."

    # 创建表
    aws dynamodb create-table \
        --table-name celeborn \
        --attribute-definitions \
            AttributeName=IPAddress,AttributeType=S \
        --key-schema \
            AttributeName=IPAddress,KeyType=HASH \
        --provisioned-throughput \
            ReadCapacityUnits=5,WriteCapacityUnits=5 \
        --region us-east-1

    if [ $? -eq 0 ]; then
        echo "Table 'celeborn' created successfully."
        LOCAL_PRIVATE_IP=$(sudo hostname -I | awk '{print $1}')
        sleep 20
        aws dynamodb put-item \
            --table-name celeborn \
            --item '{"ID": {"S": "LocalPrivateIP"}, "IPAddress": {"S": "'"$LOCAL_PRIVATE_IP"'"}}' \
            --region us-east-1
    sudo /home/ec2-user/apache-celeborn-0.5.0-bin/sbin/start-master.sh
    else
        echo "Failed to create table 'celeborn'."
        exit 1
    fi
fi

sudo yum install git gcc
git clone https://github.com/hortonworks/hive-testbench.git
wget https://dlcdn.apache.org/maven/maven-3/3.8.8/binaries/apache-maven-3.8.8-bin.tar.gz
export MAVEN_HOME=/home/hadoop/apache-maven-3.8.8
export PATH=$PATH:$MAVEN_HOME/bin
cd hive-testbench/
sudo ./tpcds-build.sh
cd /usr/bin/hive-testbench/ddl-tpcds/text
hive -d DB=‘tpcds_20000' -d LOCATION=‘s3://<your S3 bucket store the tcpds data>’ -f alltables.sql

use tpcds_20000;

spark-sql -f q77.sql —num-executors=12 —conf spark.default.parallelism=50

sudo aws s3 cp s3:/<your S3 bucket>/celeborn-client-spark-3-shaded_2.12-0.5.0.jar /usr/lib/spark/jars/

spark-sql -f q77.sql --num-executors=12 --conf spark.default.parallelism=50 --conf spark.shuffle.manager=org.apache.spark.shuffle.celeborn.SparkShuffleManager --conf spark.shuffle.service.enabled=false --conf spark.sql.adaptive.localShuffleReader.enabled=false —conf spark.celeborn.master.endpoints=<celeborn master ip>:9097