spark submit와 spark api

Spark submit이란?

: 내가 짠 코드를 Spark라는 엔진에 실어서 실행시키는 실행 버튼

 

명령어의 기본 구조

spark-submit \
  --class <메인_클래스> \
  --master <클러스터_매니저_URL> \
  --deploy-mode <배포_모드> \
  --conf <설정_키=값> \
  ... # 기타 옵션
  <애플리케이션_파일> \
  [애플리케이션_인자]

master : 스파크 작업을 어떤 환경의 자원을 빌려서 실행할 것인가

• Yarn : 하둡 사용시
  • 하둡 생태계에서 별도의 호스트 주소 없이 yarn이라고만 적으면 설정된 하둡 환경을 자동으로 찾음

• Standalone (spark://HOST:PORT) : 스파크 자체적으로 내장된 자원 관리 기능을 사용할 때 사용
  • 스파크 마스터 서버가 떠 있는 주소(HOST)와 포트(PORT)를 직접 지정해야 함. 다른 복잡한 시스템 없이 스파크만 설치된 환경에서 주로 사용
• Kubernetes (k8s://HOST:PORT) : 쿠버네티스 위에서 스파크 돌릴 때 사용
  • 쿠버네티스 API 서버 주소를 적어줌. 기본적으로 보안 연결(https)을 사용, 클러스터가 컨테이너 단위로 자원 할당
• local : 클러스터(여러 대의 컴퓨터)가 아닌, 로컬에서 딱 하나의 worker thread만 사용해 실행
  • 병렬 처리가 거의 일어나지 않아 아주 가벼움
  • local[k] : k라는 숫자만큼의 일꾼 사용
  • local[K,F] : k개의 일꾼을 사용하되, 작업 실패 시 최대 F번까지 다시 시도

 

deploy-mode : client와 cluster 두가지 버전이 있음

• 스파크 애플리케이션 실행 시 가장 중요한 두가지 프로세스는 드라이버와 엑시큐터
• 드라이버 → 지휘관, 엑시큐터 → 일꾼
• deploy-mode는 Driver이 어디서 상주하느냐를 결정
  • client mode : spark-submit 명령어를 실행한 컴퓨터(로컬)에서 드라이버 실행
    • 사용자가 실행 결과(로그)를 실시간 확인
    • 로컬 머신과 클러스터 사이의 네트워크 끊기면 드라이버 종료로 전체 중단
  • cluster mode : 드라이버가 로컬 머신이 아닌, 클러스터 내부의 작업 노드 중 하나에서 실행
    • 로컬 머신 전원 꺼도 클러스터 안에서 작업 계속 진행
    • 운영 환경에서 데이터 파이프라인 안정적으로 돌림
    • 사용자 입장에서는 "애플리케이션 매니저(Driver)를 클러스터에 던져놓고 결과 나중에 확인

pyspark 전용 설정

• --py-files : 메인 스크립트 외에 추가로 필요한 파이썬 파일, 압축된 모듈, 또는 파이썬 패키지 형태를 전달
  • 분산 환경에서 다른 노드들이 내가 만든 사용자 정의 모듈을 인식할 수 있게 배달해주는 역할
• --config.spark.executor.pyspark.memory : 각 executor 내에서 PySpark 전용으로 할당할 메모리 양을 설정
  • 스파크는 JVM 메모리를 주로 쓰지만, PySpark는 파이썬 프로세스를 따로 뛰움. 파이썬 프로세스가 데이터를 처리할 떄 메모리가 부족해 터지는 것을 방지하기 위해 이 설정으로 "파이썬용 메모리"를 확보
• --config.spark.pyspark.driver.python : 드라이버에서 사용할 파이썬 실행 파일의 경로를 지정
  • 컴퓨터나 마스터 노드에 여러 버전 파이썬이 설치되어 있을 때, 정확히 어떤 버전으로 지휘할지 정해줌
• --config.spark.pyspark.python : 드라이버와 엑시큐터 모두에서 사용할 파이썬 실행 파일 경로를 지정
  • 클러스터 환경에서는 모든 서버의 파이썬 버전이 동일해야 함. 

 

Spark API란?

: 개발자가 분사된 환경에서 복잡한 데이터 연산을 효율적으로 수행할 수 있도록 제공되는 인터페이스의 집합

• 분산 데이터 컬렉션에 대한 인터페이스 : RDD를 다루는 게 핵심으로 단일 객체를 다루듯 코드를 작성하면 API가 이를 클러스터 전체의 작업으로 변환
• 지연 평가 모델 : 즉시 실행이 아니라, 실행 계획을 세워두었다가 실제 결과가 필요한 시점에 한꺼번에 최적화해 실행

1. Transformations

• 데이터를 변형시키는 것. Spark 특성 상 데이터를 바꿀 수 없기 때문에 새로운 RDD를 만듦
• lazy해서 실행 계획에 추가함
• Actions이 트리거가 되어 Tranformations이 각각 실행

1-1. Transformation depdency : Transformations 발생 시, 부모 파티션의 데이터가 자식 파티션으로 어떻게 전달되는지

• Narrow Dependecy : 부모 파티션과 자식 파티션이 1:1 혹은 N:1로 매핑
  • n 자기 노드 안의 데이터만 읽어서 처. 다른 node와 연결을 해서 data를 셔플하지 않음. 
  • Direct data lineage : A→X, B →Y, C →Z  로 들어가는 파티션이 그대로 아웃 파티션으로 나온다
  • Efficient execution : 네트워크 통신이 없으므로 매우 빠르고 효율적
• Wide Dependency : 부모 파티션 하나가 여러 자식 파티션에 영향, 여러 부모로부터 데이터를 받아와야 함
  • node간의 데이터 셔플링 해야 한다. 
  • Data shuffling : 그래서 네트워크 I/O가 생긴다. 네트워크 트래핑이 많이 생긴다
  • Higher Cost : 데이터를 네트워크를 통해 주고 받기 때문에 비싸고 느리다. 

2. Actions

• 기록된 모든 Transformations 과정을 실행하고, 그 결과를 driver 프로그램으로 되돌려주거나 외부 저장소에 저장
• 결과로 데이터셋이 아닌 실제 값(count, list 등)을 반환하거나 파일 시스템에 기록

파티션이란?

: 데이터의 쪼개진 조각으로 대용량 데이터를 처리하기 위해 데이터를 아주 작은 조각으로 나눔

• 분산 처리의 단위 : 스파크 클러스터에는 여러 대의 서버(워커 노드)가 있는데 각 서버는 전체 데이터 중 자신에게 할당된 몇 개의 파티션만 담당해서 처리
• 물리적 실체 : 메모리나 디스크에 저장되는 실제 데이터의 블록

부모 파티션과 자식 파티션??

: 연산 전 후의 관계를 부모와 자식으로 표현

 

RDD API

1. Transformaions

1-1. Narrow dependency : map(), mapPartion(), flatMap(), filter(), union()

1-2. Wide dependency : groupByKey(), aggregateByKey(), sortByKey(), reduceByKey(), aggregate(), join(), repartitions()

2. Actions : collect(), count(), first(), take(), reduce()

 

Dataframe API

1. Transformaions

1-1. Narrow dependency : select(), filter(), withColumn(), drop(), where()

1-2. Wide dependency : groupBy(), agg(), cube(), rollup(), join(), repartitions()

2. Actions : show(), head(), collect(), count(), first()

 

RDD API가 How(어떻게)에 집중한다면 DataFrame API는 What(무엇)에 집중하고 Spark API는 이 모든 걸 가능하게 하는 시스템 전체라고 정의 가능하다.