KubeCon2025 - Spark on Kubernetes: A Practical Guide

https://www.youtube.com/watch?v=ejJ6A0sIdbw

• Spark on Kubernetes, A Practical Guide
  • Why Spark on K8s
  • Spark on Airbnb
  • Journey
    • Spark Job Submission
    • Shuffle Service
    • Observability
    • things I did't cover today (But where did we use AI?)

Damon Cortesi

• Staff SW Engineer at AirBnB
• Data Platform since 2010

Why Spark on K8s

하둡이 개발된 지 20년이 지났고 꾸준히 개발중인데 왜 k8s?

Flexibility

• Spark Version. 버전관리가 힘듬. 하둡 3에서 docker / runc Container를 지원하기 시작했으나, 이 분야로는 k8s가 훨씬 성숙하고 안정된 프로덕트.
• Compute Optimization. cluster 최적화를 위해 여러 노력을 했으나, 어떤 workload를 어떤 instance type에서 실행할 것인가? 라는 관점에서는 k8s와 Karpenter가 훌륭함.
• Resource Utilization. (Shared Workloads - in theory.)
  • 보통 spark 작업은 bursty + 수만 개의 pod가 동시에 실행되어야 함. k8s에서 잘 핸들링?
• Modernization.
  • 하둡의 java runtime 지원 속도는 느린 편. (2022년에 java 11 지원)
  • existing Spark Cluster의 버전업은 골치 아픔. 여러 버전의 cluster를 rotating하는 방식은 있지만, 버전업은 lift-and-shift 형태를 벗어나지 못함
  • Observability도 k8s에서의 지원이 더 깔끔함.
  • Developer Experience 면에서도 낫다.
• industry adoption

Spark on Airbnb

• utilization: 88%. 업계 기준 나쁘지 않은 숫자
• spark executor만 대략 90k
• 700k ~ 900k job 실행.

꽤 나쁘지 않은 효율이라고 생각함.

대부분 sparkSQL 사용.

• 쉽게 build / deploy / migrate workload할 수 있는 internal tool.
• 발표 기준, 4년 전 spark 버전을 쓰고 있음. 당시에는 k8s support가 지금처럼 잘되어 있지 않았음.
  • spark 3.11에서 지원하는 fabric8 k8s client의 경우, long lived service account token의 rotation이 지원 안 됨.
  • spark submission service 문제: spark submit하면, 실행중인 모든 job에 list watch 수행 -> not good for our cluster.
  • 그래서 backport that patch.
• Hive -> Iceberg Migration.
• 모든 spark job을 client mode -> cluster mode로 migrate.

spark submission을 사용자 노트북이나 random gateaw 등 여러 곳에서 실행하는 게 아니라 submission service 형태로 관리해서 편했음.

현재상황: 전체 workload의 38%를 k8s로 전환 완료. 지난 달에는 1% 였음.

그럼 어떻게 한 달만에 1%에서 38%로 늘었나?

• Seamless integration - 사용자는 migration 되었는지도 모른다. 아마 price 내려간거 확인한 사람들이나 짐작할 듯
• decision tree complexity <- 오늘 이야기할 부분.
  • 전환을 시도할 때, 하둡에는 있는데 k8s에서는 Not support Natively 인 것들이 있었다.
  • 하나씩 마주할 때마다 pause -> evaluate all options (license / compatiblity / etc...)
    • 크게 세 가지 관점. Cost / Performance / Human Perspective

Journey

Decision Tree Complexity의 모습. Multiple Choice Problem.

• built-in dependency들도 많음.
  • yarn에 포함된 queue / preemption / shuffle service
  • spark의 경우 하둡에 강결합된 형태로 history server를 제공한다던가..
  • spark UI의 driver link 같은 건 k8s에 없음.

이런 식의, '선택이 필요한 결정' 들을 어떻게 했는지 소개할 예정. 색칠된 것들이 특히 좀 집중했던 내용들이다.

예컨대

• 스케줄링: UniKorn 사용. yarn에 필요한 기능이 거의 다 구현되어 있었고, 지금까지도 잘 동작중.

Spark Job Submission

Spark Job != K8s Job.

• 대충 batchJob 형태니까 대체 가능한 거 아님? No. 내부 동작이 다름

Spark Submit 실행하면

• create driver PodSpec -> submit to k8s cluster. 일종의 kubectl apply
• spark driver 자체도 pod 생성권한이 필요하다. 즉 실제로 pod의 lifecycle을 관리하는 건 spark driver.
• 그래서, 일반 사용자가 직접 spark driver 제어권한을 갖게 되면, 모두가 pod 통제권한을 갖게 된다.

그래서 가능한 대안이 크게 두 개. KubeFlow Operator / Apache Operator.

• kubeflow operator
  • google이 개발했고, 작년 초 (2024년)에 kubeflow 재단으로 넘어감.
  • JVM의 spark-submit 프로세스를 fork한 방식.
  • 일반적인 사용패턴인 single operator pod 실행 -> JVM이 cpu / memory를 많이 점유하는걸 보게 된다. (slow down pretty quickly)
• Apache Operator
  • 2025년에 release.
  • Java의 Spark library 바로 호출하므로, 상대적으로 빠름
  • Prometheus Metric becomes Flatten - 쿼리가 불편함
  • SparkCluster 지원
    • standalone cluster for local development 용도였는데, k8s에서 띄울 수 있게 마개조된 프로덕트
    • vCluster 생성, statefulset for scale, has Own UI...

근데 우린 둘 다 안쓰기로 함. Adding the Operators Just made it harder.

• 우린 자체 job Submission Service가 있었음. operator를 넣었을 때, 우리 쪽 로직이랑 뭐가 계속 충돌함
  • i.e. 자체 submission에서 만들어진 ID와 operator가 만든 ID가 있는데, operator의 id는 retry 과정에서 increment. 이게 누적되니까 spark app tracing하기 힘들다.
  • (나중에 apache operator에서 increment 로직 수정됐더라)
• 여튼, k8s의 특정 namespace에 submit driver한다! 에서 시작하자. Operator 이슈는 뒤에서 또 이야기가 나온다.

Shuffle Service

DRA까지 생각한다면, k8s 환경에서 shuffle을 적용하는 건 굉장히 머리아픈 작업이다.

• task 양에 따라 spark가 workload를 scale up / down 하는 것
• spark에서 이 기능을 가능하게 해주는 shuffle Service는 k8s 컴포넌트에 대응되는 게 없다.
  • k8s에서 이 기능을 활성화하려면 shuffle tracking 옵션 활성화 or Remote Shuffle Tracking 기능이 필요.
  • shuffle tracking이 켜지면?
    • A executor의 데이터 처리 결과를 B executor에서 필요로 할 경우, A executor는 데이터 작업이 끝났음에도 pod가 죽지 않고 대기한다.
    • efficiency 측면에서는 나쁜 선택지.

그러면 Remote Shuffle Service을 이용해야 함. 외부 솔루션들을 평가해보면

• Celeborn
  • Better Docs
  • Configuration Option이 아주 많아서, 열심히 읽어야 함
  • 팀에서 쓰는 Spark 3.11 버전 Support도 되었기에 최종 선택
• Uniffle
  • Docs 별로임. 소스코드 봐야 할 일이 많았음
• Uber RSS: Not maintained. / AWS Cloud Shuffle: Bad Performance

Shuffle Performance 테스트. (TPC-DS Query Performance - 3TB scale)

TL;DR : Make sure your discs are optimized.

• 0%: current Spark Clusters (GP3 EBS)
• Stock k8s Environment (Yellow line)
  • Spark on k8s가 전반적으로 Performs Better.
• k8s NVMe disk (Red line)
  • 피크 튀는 지점: TPC-DS queries that are shuffieheavy. (disk heavy). obviously, if you're running an EBS with GP2 disk under the hood.

같은 인스턴스 타입, 같은 spark version으로 진행했음에도 spark on k8s가 성능 면에서는 더 나았다.

shuffle service가 적용된 클러스터의 모습. celeborn을 k8s 클러스터 내부에 배포했음.

• easy up / running. statefulset 형태로 배포하며, Spark Executors connect directly to each of celeborn workers.
• use clusterLocalDNS in addressing.

이 구조도 나중에 다시 조명할 예정.

Observability

세 개의 컴포넌트가 필요함

• Spark UI: running / completed job 상태 표시.
  • Running Job 데이터?
    • runs on Driver Pod. local port 노출하고 Spark Driver가 붙는 형태.
    • 단, 모든 Spark driver가 붙을 순 없음. 하루에 실행되는 job만 70k인데, 이게 전부 mesh cluster에 붙으면 안되기 때문
  • Completed Job
    • spark는 event-based json 파일을 생성. 작업이 끝나면 파일 upload -> spark history server에서 read / reconstruct the state.
• Container logs & Spark Event log
  • stdout, stderr.
  • spark 4.0 부터 structured logging을 지원함. (spark 3은 X)
  • 따라서 사용자가 찍은 print / debugging 로그... 등등을 standard log format & forwarding해야 함.
• State Tracking

Operator 방식 - CRD 정의하고, CR에 기록해주는 구조가 있긴 한데 안 썼음. (Apache Operator에 버그가 있다는 듯)

Container Log용 선택지들. 요구사항은 아래와 같다

• Spark UI와 Linkable (spark의 stderr, stdout 지원 + HTTP 형태로 외부에서 접근 가능)
• Low Latency (second 단위)
• Maintain Ordering
• ~ 90TiB / Week (for spark Containers)

고민했던 선택지들

• fluentbit
• vector
• airbnb 자체 logging pipeline
• Otel

최종적으로는 Otel 선택

• Metric Collection으로 사내에서 쓰고 있음. Few moving components as possible.

전체 노드에 Daemonset 설치하는 대신 spark driver / executor에 collector sidecar 붙이는 식으로 도입.

• stdout, stderr를 s3에 전송
• 로그파일에 uuid v7 timestamp (time-sortable uuid) 추가하는 정도의 변경.
  • sidecar가 모종의 이유로 restart되더라도 reconstitute.
  • fluentbit은 state 관리를 incremental integer로 하던데, otel에서는 fluent하게 지원해주지 않음

User experience를 위해 Per-Cluster Reverse Proxy 제공.

• 사용자가 spark UI / history server / logs 전부 접근할 수 있도록.
  • /ui : 이름이나 label 기준으로 running pod 조회 가능
  • /shs: 작업이 끝난 pod 조회 시 redirect되는 곳. (history server endpoint)
    • 개인적으로 UI에서 running job 보고있었는데 작업 끝나면 history server 따로 가서 보는게 불편했음
  • /logs: jobID 기준으로 조회하면 s3 로그를 concatenate -> end user에게 제공.

• 추가로 history server 커스텀한 부분
  • 원래 history server는 계속 application log을 load하는 방식.
  • 우리는 workload가 많아서 scanning이 default option이면 불필요한 부하가 들어간다 = on-demand 방식으로 변경. (save S3 cost)

State Tracking

• Operator는 spark Job 상태확인을 위해 CR을 조회. job 상태가 바뀔 때마다 CR의 필드를 수정하는 구조.
  • 하지만 우리는 그 정도로 k8s polling하고 싶지 않음.
• base reconciler를 구현한 방식
  • each spark namespace에 reconciler 배포.
    • driver가 실행되면 monitor the state of spark driver -> push the state back to our Spark submission server. (real time update)
  • finalizer on the driver pods 구현
    • driver 상태를 확인하기 전에 k8s에서 driver pod를 GC하면 안 됨.
    • submission server에 성공적으로 state를 전부 전달했을 때 remove the finalizer -> k8s에서 자동 GC.
• 이 구현으로도 etcd 부하를 줄이는 데 한계가 있어서, Custom De-scheduler 구현. (completed pod 있으면 GC)

Observability 까지 포함된 infrastructure.

• 사용자가 spark workload 배포 - Driver / Executor 생성
  • sidecar 붙어서 로그 전송.
  • status reporter가 상태 전달
• Portal로 사용자가 job id 확인할 수 있음.
  • 어느 클러스터 (기존 spark / new k8s)에 배포되건 portal에서 id로 조회할 수 있다.
  • = 사용자에게 영향 없는 migration

추가로, 'Airflow DAGs에서 특정 조건을 만족하는 것들은 k8s에서 실행 / 동작 실패 시 old Environment에서 실행' 하는 식으로 처리했음. (모든 job은 idempotent)

배포는?

• spinnaker Triggers argoCD.
  • App on Apps 패턴
• 모든 config (컴포넌트)는 Git에 저장
  • 오픈소스 helm chart, internal helm chart, 필요한 시스템 컴포넌트...
• 여러 개의 cluster / 여러 개의 namespace에 배포.

things I did't cover today (But where did we use AI?)

)

요새 AI 안넣으면 발표 못하잖아 (ㅋㅋ)

• Kubeflow Spark History Server의 MCP.
  • 최근에 오픈소스로 나왔음. 사용자가 직접 디버깅할 때 만족도가 '매우' 높음.
• scheduling
  • YuniKorn, Volcano, Kueue 관련해서 Thorough evaluation 했는데 소개를 못하네. 아쉽.
• 매일 보고있는, 개쩌는 Grafana Dashboard
• scaling etcd / ip addresses
  • etcd limit to 16G / scale up node size
  • api server node, pre-warming IP addresses...

다음 kubeCon에서 Accepted되면 할 수 있을 듯.

관련해서 비슷한 경험을 하고 있다면 연락달라. Community 만들고 싶음.