[aws] emr.sh

# emr.sh

#!/bin/sh

TAG=$1

ENV=$2
CLASS=$3
ARGS=$4

SRC=s3://src/${ENV}/jar/batch/batch.jar
LOG=s3://log/${ENV}/batch/

SUBNET_ID=subnet-0e3653577617c98a3

echo $( \
aws emr create-cluster \
\
--auto-scaling-role EMR_AutoScaling_DefaultRole \
--instance-groups file://./batch/static/json/instance.json \
\
--name ${CLASS} \
--release-label emr-6.7.0 \
--auto-terminate \
--applications Name=Spark \
--use-default-roles \
--ec2-attributes SubnetId=${SUBNET_ID} \
--tags \
Env=${ENV} \
Name=${TAG} \
--log-uri ${LOG} \
--configurations file://./batch/static/json/spark.json \
\
--steps "Type=SPARK,Name=${CLASS},ActionOnFailure=TERMINATE_CLUSTER,Args=[--class,${CLASS},${SRC},${ARGS}]" \
)

# spark.json

[
  {
    "Classification": "spark-env",
    "Configurations": [
      {
        "Classification": "export",
        "Properties": {
          "JAVA_HOME": "/usr/lib/jvm/java-11-amazon-corretto.x86_64"
        }
      }
    ]
  },
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.executorEnv.JAVA_HOME": "/usr/lib/jvm/java-11-amazon-corretto.x86_64",
      "spark.sql.broadcastTimeout": "3600",
      "spark.default.parallelism": "200",
      "spark.yarn.am.memory": "2g",
      "spark.executor.extraJavaOptions": "-XX:+IgnoreUnrecognizedVMOptions",
      "spark.rpc.askTimeout": "600s",
      "spark.sql.shuffle.partitions": "360",
      "spark.sql.cbo.enabled": "True",
      "spark.sql.adaptive.enabled": "True",
      "spark.sql.adaptive.coalescePartitions.enabled": "True",
      "spark.sql.adaptive.advisoryPartitionSizeInBytes": "128m",
      "spark.dynamicAllocation.enabled": "True",
      "spark.dynamicAllocation.initialExecutors": "1",
      "spark.dynamicAllocation.minExecutors": "1",
      "spark.dynamicAllocation.maxExecutors": "300",
      "spark.dynamicAllocation.executorAllocationRatio": "1",
      "spark.sql.catalogImplementation": "hive",
      "spark.hadoop.hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory",
      "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
      "spark.kryoserializer.buffer.max": "1024m",
      "spark.sql.autoBroadcastJoinThreshold": "60mb"
    }
  },
  {
    "Classification": "hive-site",
    "Properties": {
      "hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
    }
  },
  {
    "Classification": "spark-hive-site",
    "Properties": {
      "hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
    }
  },
  {
    "Classification": "spark",
    "Properties": {
      "maximizeResourceAllocation": "true"
    }
  }
]

# instance.json

[
  {
    "InstanceCount": 1,
    "Name": "MASTER",
    "InstanceGroupType": "MASTER",
    "InstanceType": "r4.xlarge",
    "BidPrice": "0.1"
  },
  {
    "InstanceCount": 1,
    "Name": "CORE",
    "InstanceGroupType": "CORE",
    "InstanceType": "r4.xlarge",
    "BidPrice": "0.1",
    "AutoScalingPolicy": {
      "Constraints": {
        "MinCapacity": 1,
        "MaxCapacity": 100
      },
      "Rules": [
        {
          "Name": "Default-scale-out",
          "Description": "Replicates the default scale-out rule in the console for YARN memory.",
          "Action": {
            "SimpleScalingPolicyConfiguration": {
              "AdjustmentType": "CHANGE_IN_CAPACITY",
              "ScalingAdjustment": 1,
              "CoolDown": 300
            }
          },
          "Trigger": {
            "CloudWatchAlarmDefinition": {
              "ComparisonOperator": "LESS_THAN",
              "EvaluationPeriods": 1,
              "MetricName": "YARNMemoryAvailablePercentage",
              "Namespace": "AWS/ElasticMapReduce",
              "Period": 300,
              "Threshold": 15,
              "Statistic": "AVERAGE",
              "Unit": "PERCENT",
              "Dimensions": [
                {
                  "Key": "JobFlowId",
                  "Value": "${emr.clusterId}"
                }
              ]
            }
          }
        }
      ]
    }
  }
]

FROM python:3.10.6 as app

ENV TZ=Asia/Seoul

RUN pip install awscli

WORKDIR /app
COPY static/ /app/static/

export AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
export AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
export AWS_DEFAULT_REGION="ap-northeast-2"

TAG="test-emr"

ENV=prd
CLASS=test

output=$( \
./batch/static/sh/emr.sh \
${TAG} \
\
${ENV} \
${CLASS} \
"${TAG},${AWS_ACCESS_KEY_ID},${AWS_SECRET_ACCESS_KEY},${AWS_DEFAULT_REGION},${BUCKET_NAME},${S3_PATH},${MAIL},${TSV_FIELDS},\"${QUERY}\"" \
)


curl -L https://github.com/stedolan/jq/releases/download/jq-1.6/jq-linux64 -o ./jq
chmod a+x ./jq
cluster=$(echo ${output} | ./jq ".ClusterId")
cluster=$(echo ${cluster} | sed -e 's/"//g')

echo ${cluster}

https://huzz.tistory.com/19

AWS CLI로 EMR Spark Cluster 띄우기

https://docs.aws.amazon.com/ko_kr/emr/latest/ReleaseGuide/emr-spark-configure.html#spark-change-defaults

Spark 구성 - Amazon EMR

https://stackoverflow.com/questions/70886684/how-to-use-java-runtime-11-in-emr-cluster-aws

How to use java runtime 11 in EMR cluster AWS

https://stackoverflow.com/questions/62928662/facing-error-while-trying-to-create-transient-cluster-on-aws-emr-to-run-python-s

Facing error while trying to create transient cluster on AWS emr to run Python script

https://docs.aws.amazon.com/ko_kr/emr/latest/ReleaseGuide/emr-configure-apps.html

애플리케이션 구성 - Amazon EMR

https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-configure.html

Configure Spark - Amazon EMR

https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-automatic-scaling.html

Using automatic scaling with a custom policy for instance groups - Amazon EMR

https://github.com/WorksApplications/ansible_aws_emr/blob/0b58e7223de36ed35b89b4b92c9391431315b27d/emr/examples/roles/emr/init-create-fleet/files/emr_config.json

GitHub - WorksApplications/ansible_aws_emr: Unofficial Ansible module for Amazon EMR

https://gist.github.com/tmusabbir/34fdab6bd30fd87bcdd69cf03f54090c

AWS CLI command to create EMR cluster with default auto-scaling task group