📒 Spark(15)

📌 Broadcast Variable

Broadcast Variable이란

• 룩업 테이블등을 Broadcasting하여 Shuffling을 막는 방식으로 사용한다.

  • Broadcast Join에서 사용되는 것과 동일한 테크닉이다.

  • 대부분 룩업 테이블(혹은 디멘션 테이블 - 10~20MB)을 Executor로 전송하는데 사용한다.
    -> 많은 DB에서 스타 스키마 형태로 팩트 테이블과 디멘션 테이블을 분리한다.

  • spark.sparkContext.broadcast를 사용한다.

LookUP Table - LUT

• 주어진 연산에 대해서 미리 계산된 결과들의 집합(배열)을 가리킨다.

• 이 집합(배열)은 주어진 연산에 대한 결과를 계산하는 시간보다 더 빠르게 값을 취득해 갈 수 있도록 사용되는 레퍼런스로 사용된다.

• 컴퓨터 과학에서 일반적으로 배열이나 연관 배열로 된 데이터 구조를 LUT라 부른다.

룩업 테이블(파일)을 UDF로 보내는 방법

• Closure

  • Serialization이 Task단위로 일어난다.
  • UDF안에서 Python Data Structure를 사용하는 경우

• Broadcast

  • Serialization이 Worker Node단위로 일어난다.(그 안에서 캐싱되기에 훨씬 더 효율적이다.)
  • UDF안에서 Broadcast된 Data Structure를 사용하는 경우

• Broadcast Dataset의 특징

  • Worker node로 공유되는 변경 불가 데이터
  • Worker node별로 한 번 공유되고 캐싱된다.
  • 제약적으로는 Task Memory안에 들어갈 수 있어야한다는 점이다

---

📌 Accumulators

Accumulators란

• 특정 이벤드의 수를 기록하는데 사용된다. -> 일종의 전역 변수
  -> 하둡에서 카운터와 아주 흡사하다.

• 예를 들면 비정상적인 값을 가지는 레코드의 수를 카운팅하는데 사용된다.

def handle_bad_zipcode(c:int) -> int:
	if len(str(c)) != 5:
    	return None
    return c

Accumulators의 특징

• 변경 가능한 전역변수로 드라이버에 위치한다.

• 스칼라로 만들면 이름을 줄 수 있지만 그이외에는 불가능하다.
  -> 이름있는 Accumulators만 Spark Web UI에 나타난다.

• 레코드 별로 세거나 합을 구하는데 사용 가능하다.

• 두 가지 방법으로 사용 가능하며 값의 정확도도 달라진다.

  • Transformation에서 사용한다.
    -> 이 경우 값이 부정확할 수 있다.(Task의 재실행과 Speculative Execution)
  • DataFrame/RDD Foreach에서 사용한다.
    -> 추천되는 방식으로 이 경우 정확하다.

---

📌 Speculative Execution

Speculative Execution란

• 느린 Task를 다른 Worker node에 있는 Executor에서 중복 실행한다.
  • 이를 통해 Worker node의 하드웨어 이슈등으로 느려지는 경우 빠른 실행을 보장한다.
  • 하지만 Data Skew로 인해 오래 걸린다면 도움이 안되고 리소스만 낭비하게 된다.
  • 다른 Worker node로 새로운 Task로 만들고 먼저 실행되는 것을 사용하고 나머지는 Kill 한다.

Speculative Execution 제어방식

• spark.speculation으로 컨트롤 가능하며 기본은 False이다.
  -> 하둡 MapReduce에서부터 있던 기능이다.
• 다양한 환경 변수로 세밀하게 제어 가능하다.

환경 변수 이름	Default	LinkedIn
spark.speculation.interval	100ms	1 sec
spark.speculation.multiplier	1.5	4
spark.speculation.quantile	0.75	0.9
spark.speculation.minTaskRuntime	100ms	30 sec
spark.speculation.task.duration.threshold	None	None

Spark의 리소스 할당(스케쥴링)

• Spark Application들간의 리소스 할당

  • 기반이 되는 리소스 매니저가 결정된다.
    -> YARN은 세 가지 방식을 지원한다. FIFO, FAIR, CAPACITY
  • 한 번 리소스를 할당 받으면 해당 리소스를 끝까지 들고 가는 것이 기본이다.

• 하나의 Spark Application안에서 Job들 간의 리소스 할당

  • FIFO 형태로 처음 Job이 필요한 대로 리소스를 받아서 쓰는 것이 기본이다.

Spark Application의 리소스 요구/릴리스 방식

• Static Allocation(기본 동작)

  • Spark Application은 리소스 매니저로부터 (YARN)받은 리소스를 보통 끝까지 들고간다.
  • 이는 리소스 사용률에 악영향을 줄 가능성이 높다.

• Dynamic Allocation

  • Spark Application이 상황에 따라 executor를 릴리스하기도 하고 요구하기도 한다.
  • 다수의 Spark Application들이 하나의 리소스 매니저를 공유한다면 활성화하는 것이 좋다.

spark-submit --num-executors 100 --executor-cores 4 --executor-memory 32G

---

📌 Scheduler

Scheduler란

• 하나의 Spark Application내의 Job들에 리소스를 나누어주는 정책이다.
  -> Spark Application들간에 리소스를 나누어주는 방식은 리소스 매니저에게 달려있다.

• 다음 2가지가 존재한다.

  • FIFO(기본)

    • 리소스를 처음 요청한 Job에게 리소스 우선 순위가 간다.

  • FAIR

    • 라운드로빈 방식으로 모든 Job에게 고르게 리소스를 분배하는 방식이다.
    • 이 안에서 Pool이란 형태로 리소스를 나누어서 우선순위를 고려한 형태로 사용 가능하다.
      -> Pool안에서 리소스 분배도 FAIR 혹은 FIFO로 지정 가능하다.

Scheduler를 활용한 병렬성 증대

• 병렬성 증대 -> Thread 활용이 필요하다.

  • 이는 FAIR 모드의 스케쥴러일 경우 더 효과적이다.

• 관련 환경 변수

  • spark.scheduler.mode : FIFO(기본) or FAIR
  • spark.scheduler.allocation.file: "FAIR"인 경우 필요하며 Pool을 정의해놓는 형태로 사용된다.

---

📌 Dynamic Resource Allocation

• 아래 환경변수들로 제어가 가능하다.
• spark.dynamicAllocation.enabled = tru
• spark.dynamicAllocation.shuffleTracking.enabled = true
• spark.dynamicAllocation.executorldleTimeout = 60s(릴리스 타이밍 결정)
• spark.dynamicAllocation.schedulerBacklogTimeout = 1s
• spark.dynamicAllocation.minExecutors
• spark.dynamicAllocation.maxExecutors
• spark.dynamicAllocation.initialExecutors
• spark.dynamicAllocation.executorAllocationRatio

참고