[Kubernetes] Amazon EMR on EKS 운영을 위한 기본 지식

Work/개발 노트

[Kubernetes] Amazon EMR on EKS 운영을 위한 기본 지식

★용호★ 2023. 11. 12. 02:04

이 게시글은 학습을 위해 정리한 내용이므로 잘못된 내용이 있을 수 있습니다. 잘못된 정보가 있다면 댓글로 남겨주세요. 수정하겠습니다.

용어 정리

스파크 애플리케이션 : 컴파일된 JAR나 라이브러리 파일을 의미. submit하는 시점에 로컬 머신에서 코드가 실행되어 클러스터 드라이버 노드에 요청함
드라이버 프로세스 : SparkContext를 생성하고, 자신이 담당한 Task가 클러스터 내에서 실행될 수 있도록 하며 진행 상태 모니터링
익스큐터 프로세스 : 드라이버 프로세스가 할당한 작업 수행 (각 스파크 애플리케이션 마다 개별 프로세스 사용)
클러스터 매니저 : 물리적 머신을 관리하며, Spark 애플리케이션 실행에 필요한 자원을 할당
Spark Session : SparkSession은 SparkContext를 안전하게 생성하도록 도와주는 빌더 패턴 라이브러리
DataFrame : 테이블의 데이터를 로우와 컬럼으로 단순하게 표현 (가장 대표적인 구조적 API)
파티션 : 클러스터의 물리적 머신에 존재하는 로우의 집합을 의미
트랜스포메이션 : DataFrame을 변경하기 위해 변경 방법을 스파크에 알려주기 위한 명령
스파크 UI : 스파크 Job의 진행 상황을 모니터링할 때 사용

Spark Job 실행 로직

Driver 프로세스가 main 함수를 실행하고, Spark Context를 생성
Spark Context가 Cluster Manager에 연결
Cluster manager로부터 Executor 할당 받음
Spark Context가 Jar 또는 Python 파일을 Executor에 전달
SparkContext가 Task를 Executor에 전달
Executor가 Task를 모두 수행하면서 진행 상황과 결과 보고

Spark 애플리케이션에서 Job 실행 Flow

드라이버와 익스큐터의 실행 리소스 정보를 포함하여 스파크 애플리케이션 Submit
클러스터 매니저가 클러스터 노드 중 하나에 드라이버 프로세스 실행
사용자 코드에 포함된 SparkSession 초기화 로직으로 클러스터 매니저와 연결
SparkSession은 클러스터 매니저에게 익스큐터 프로세스의 실행 요청 (spark-submit 실행 시 사용된 명령행 인수에 관련 설정이 포함됨)
익스큐터 프로세스가 실행되면 클러스터 매니저는 드라이버 프로세스로 익스큐터의 위치와 관련된 정보를 전달
드라이버와 워커는 코드를 실행하여 데이터를 이동하는 과정을 서로 통신함
드라이버는 각 워커에 태스크를 할당
태스크를 받은 워커는 태스크의 상태 정보를 드라이버에 전송
스파크 애플리케이션 실행이 완료되면 드라이버 프로세스가 성공/실패 중 하나의 상태로 종료되고 해당 정보를 드라이버로 전송
클러스터 매니저는 드라이버가 속한 스파크 클러스터의 모든 익스큐터를 종료시킴

Kubernetes 상에서 Spark 애플리케이션 실행 유형

Why EMR on EKS?

아래 조건에 해당하는 사용자에 적합

기존에 Amazon EKS 클러스터를 운영하고 있던 경우
기존 EMR on EC2 또는 Spark on EKS로 운영하고 있던 경우

이점

Spark용 EMR 런타임은 오픈 소스 Apache Spark와 100% API 호환되는 Spark용 성능 최적화 런타임
Arm 기반 Gravition 인스턴스 지원 (성능 최대 15% 향상, 비용 최대 30% 절감)
디버깅 및 모니터링 옵션 제공
멀티 테넌트 요건 충족
Job 실행에 IAM 역할 부여로 애플리케이션 수준 보안 제어
EMR on EKS에서 보안 및 성능 패치를 지원하기 때문에 애플리케이션에 집중
컴퓨팅을 분리하여 단일 Amazon EKS 클러스터에서 동일한 런티임의 다양한 버전 및 설정으로 실행할 수 있음
Amazon EMR Studio를 통해 데이터 과학자 및 개발자를 위한 환경 제공
AWS 서비스와의 통합

Spark on EKS vs EMR on EKS

3TB 규모의 TPC-DS 데이터 세트(177억개 레코드, Parquet 파일)를 사용한 벤치마크 테스트에서 EKS 상에 오픈소스 Spark를 실행하는 것보다 최대 61% 더 낮은 비용과 최대 68% 향상된 성능 제공 (참고)

TPC-DS(Transaction Processing Performance Council-Decision Support) : 빅데이터 기술의 분석 성능을 평가하는데 사용되는 의사 결정 지원 벤치마크, DW 영역에서 널리 사용되는 테스트 데이터 셋 제공

우아한형제들 사례 - Spark on EKS로 운영 시 Spark 버전 업데이트 관리 문제로 EMR on EKS 선택
- EMR 클러스터 대비 빠른 프로비저닝 시간 소요(15분 → 2분)
- master, core 등 상시 운영 노드가 필요 없음
- 단일 클러스터 내에서 여러 Spark 버전을 쉽게 사용 가능
- Karpenter를 통해 노드 오토스케일링을 더 빠르게 수행
- IAM 기반 사용자 권한 제어를 쉽게 적용 가능

노드 관리

Karpenter 기반의 노드 확장

Spark 워크로드에 따라 리소스 요구사항이 다를 수 있어서 ASG 기반의 CAS 보다는 인스턴스 유형을 유연하게 선택할 수 있는 Karpenter가 유리 (EC2 Fleet API를 직접 호출해서 인스턴스 프로비저닝)

여러 AZ에 걸쳐서 Driver와 Executor Pod가 실행될 수 있고 이 경우 AZ간 네트워크 I/O로 인한 성능 저하가 발생할 수 있기 때문에 단일 AZ를 활용하는 것이 성능 및 비용에 유리

apiVersion: karpenter.sh/v1alpha5
kind: Provisioner
metadata:
  name: default
spec:
  labels:
    app: demo
    
  requirements:
    - key: "karpenter.sh/capacity-type"
      operator: In
      values: ["on-demand","spot"]
    - key: "kubernetes.io/arch" 
      operator: In
      values: ["amd64"]
    - key: karpenter.k8s.aws/instance-family
      operator: In
      values: [c5, c5a, c5d, c5ad, m5, c6a]
    - key: karpenter.k8s.aws/instance-size
      operator: In
      values: [2xlarge, 4xlarge, 8xlarge, 9xlarge]
    - key: "topology.kubernetes.io/zone"
      operator: In
      values: ["{AWS_REGION}a"]

  ttlSecondsAfterEmpty: 30

labels를 사용하여 해당 Provisioner로 생성되는 모든 노드에 label 지정
- spark submit 실행 시 특정 노드에 Pod들을 실행하도록 할 수 있음
Executor Pod는 중단 시 재실행 될 수 있으므로 Spot 인스턴스를 활용하는 것이 비용 효율적
재작업으로 인한 지연을 최소화하기 위해 다양한 인스턴스 유형을 지정하여 Spot 인스턴스 고갈 시 회수 방지

EMR의 Pod 템플릿을 사용하여 Driver Pod는 온디맨드, Executor Pod는 Spot으로 실행하기 위해 Pod 템플릿 활용

apiVersion: v1
kind: Pod
spec:
  nodeSelector:
    karpenter.sh/capacity-type: spot
  containers:
  - name: spark-kubernetes-executor


apiVersion: v1
kind: Pod
spec:
  nodeSelector:
    karpenter.sh/capacity-type: on-demand
  containers:
  - name: spark-kubernetes-driver

spark submit 실행 시 Karpenter로 실행 될 수 있도록 설정

aws emr-containers start-job-run \
  --virtual-cluster-id $VIRTUAL_CLUSTER_ID \
  --name karpenter-demo  \
  --execution-role-arn $EMR_ROLE_ARN \
  --release-label emr-6.5.0-latest \
  --job-driver '{
  "sparkSubmitJobDriver": {
      "entryPoint": "local:///usr/lib/spark/examples/jars/eks-spark-benchmark-assembly-1.0.jar",
      "entryPointArguments":["s3://blogpost-sparkoneks-us-east-1/blog/BLOG_TPCDS-TEST-3T-partitioned","s3://'$S3BUCKET'/EMRONEKS_TPCDS-TEST-3T-RESULT-KA","/opt/tpcds-kit/tools","parquet","3000","1","false","q70-v2.4,q82-v2.4,q64-v2.4","true"],
      "sparkSubmitParameters": "--class com.amazonaws.eks.tpcds.BenchmarkSQL --conf spark.executor.instances=50 --conf spark.driver.cores='$CORES' --conf spark.driver.memory='$EXEC_MEMORY'g --conf spark.executor.cores='$CORES' --conf spark.executor.memory='$EXEC_MEMORY'g"}}' \
  --configuration-overrides '{
    "applicationConfiguration": [
      {
        "classification": "spark-defaults", 
        "properties": {
          "spark.kubernetes.node.selector.app": "demo",
          "spark.kubernetes.node.selector.topology.kubernetes.io/zone": "'${AWS_REGION}'a",

          "spark.kubernetes.container.image":  "'$ECR_URL'/eks-spark-benchmark:emr6.5",
          "spark.kubernetes.driver.podTemplateFile": "s3://'$S3BUCKET'/pod-template/karpenter-driver-pod-template.yaml",
          "spark.kubernetes.executor.podTemplateFile": "s3://'$S3BUCKET'/pod-template/karpenter-executor-pod-template.yaml",
          ...생략...
      }
    }

Spark는 디스크에 중간 데이터를 읽고 쓰는 데이터 셔플 동작이 있고, 여기에 리소스가 많이 필요하기 때문에 스토리지 최적화 인스턴스를 사용하는 것이 유리
- 성능이 좋지 않은 디스크가 있으면 작업의 전체 성능이 저하되고, 디스크가 가득 차면 작업이 실패함
- NVME 인스턴스 저장소가 있는 Amazon EC2 노드를 사용하고 디스크를 Spark Pod에 HostPath 볼륨으로 탑재
- 컴퓨팅 제품군 중 높은 디스크 I/O 성능을 제공하는 d 시리즈의 인스턴스 유형을 사용 권장 (ex c5d.9xlarge)

--conf spark.kubernetes.executor.volumes.hostPath.spark-local-dir-1.mount.path='/tmp/spark' \
--conf spark.kubernetes.executor.volumes.hostPath.spark-local-dir-1.options.path=/pv-disks/local \
--conf spark.local.dir='/tmp/spark' \

Arm 기반 Gravition 인스턴스 활용 (성능 최대 15% 향상, 비용 최대 30% 절감)

EMR on EKS 동작 방식

Spark Job 정의에 각 애플리케이션 별로 파라미터를 포함하여 Amazon EMR로 Job을 submit
Amazon EMR은 파라미터를 확인하여 배포할 컨테이너 정보를 기반으로 Amazon EKS에 Pod를 실행할 것을 지시함
Amazon EKS는 Job을 실행하는데 필요한 Pod를 EC2 인스턴스 또는 Fargate 상에 실행하여 Job 실행

이런 느슨한 서비스 결합으로 안전하게 격리된 여러 작업을 동시에 실행할 수 있음

비용 모니터링

Kubecost로 EMR on EKS로 사용된 컴퓨팅 사용량으로 인한 비용 증가를 파악하고, AWS CUR을 통해 컴퓨팅 비용 조정
Karpenter를 사용하여 EC2 리소스 활용도 최적화 (유휴 리소스 비용 감소)
executor와 driver Pod에는 아래 두 label이 포함되므로 Kubecost를 사용하여 추적
- emr-containers.amazonaws.com/job.id
- emr-containers.amazonaws.com/virtual-cluster-id