spark 고급,spark ML - 2

이영준·2024년 7월 10일

big data, spark, hadoop

목록 보기

7/9

Cashing and Persist

어느 시스템이건, 반복되어서 사용하는 데이터가 있다면 메모리에 두는 것이 좋음
그 방법이 spark에서의 사용하는 것이 Cashnig과 Persist

Cashing

storage Memory pool에 할당이 되는데, 모든 데이터 프레임을 여기에 cashing을 할 수 없음.
Cashing이란 무엇이며, 왜 cashing이 필요한가?
- 반복되는 데이터를 메모리에 둠으로써 재사용
어떻게 DataFrame을 cashing할 것인가?
- cashing과 Persist인데 여기선 cashing만 다룸
언제 cashing하고 언제 하지 말아야 하는가?
cashing을 취소하는 방법
cashing 포맷
cashing을 메모리에 할까 디스크에 할까(필요하다면)

cashing이란

자주 사용되는 데이터 프레임을 메모리에 유지하여 처리 속도 증가.
- 단 그 데이터 프레임이 정말 메모리에 있는지 확인 필요
- 어떤 경우에는 다시 계산하는 것이 빠를 수 있음.
메모리 소비를 늘리므로 불필요하게 모든 걸 캐싱할 필요는 없음.
개인적인 이야기로 아두이노의 EEPROM이 캐싱할 때 개념 이해가 확 됐음

어떻게 DataFrame을 caching하는가?

두 가지 방법
- cache()
- persist()
둘 다 모두 데이터프레임을 메모리/디스크/오프힙에 보존
- 모두 lazy execution - 필요해지기 전까지 캐싱하지 않음.
- caching은 항상 파티션 단위로 메모리에 보존
  - 하나의 파티션이 부분적으로 caching 되지 않음.

Spark SQL - caching

spark.sql("cache table table_name")
spark.sql("cache lazy table table_name")
- 이건 쓰일때 메모리에 쓰임.
spark.sql(""uncache table table_name)

spark SQL - caching 취소

DataFrame.unpersist (LRU - Least Recently Used)
spark.sql("uncache table table_name")
spark.catalog.isCached("table_name")
spark.catalog.clearCache()

Caching 실습

실습

이 코드는 PySpark를 사용하여 데이터 프레임을 생성하고, 캐싱을 통해 성능을 최적화하는 과정을 보여줍니다. 이를 요약 및 분석하여 설명하면 다음과 같습니다.

df = spark.range(1, 1000000).toDF("id"):
- 1부터 999,999까지의 숫자를 갖는 데이터 프레임 df를 생성함.
df.show(5):
- 데이터 프레임 df의 처음 5개 행을 출력함.
df10 = df.repartition(10):
- 데이터 프레임 df를 10개의 파티션으로 나눔.
df10_square = df10.withColumn("square", expr("id*id")):
- id 열의 값을 제곱한 새로운 열 square를 추가하여 데이터 프레임 df10_square를 생성함.
df10_square.cache():
- 데이터 프레임 df10_square를 캐시에 저장함.
df10_square.take(10):
- 데이터 프레임 df10_square의 처음 10개 행을 가져옴.
- 이 과정에서 첫 번째 파티션이 메모리로 로딩됨. 이는 Spark Web UI에서 확인할 수 있음.
df10_square.count():
- 데이터 프레임 df10_square의 전체 행 수를 셈.
- 이 과정에서 모든 파티션이 메모리로 로딩됨. 이는 Spark Web UI에서 확인할 수 있음.
df10_square.unpersist():
- 데이터 프레임 df10_square를 캐시에서 제거함.
df10_square.createOrReplaceTempView("df10_square"):
- 데이터 프레임 df10_square를 임시 뷰로 생성함.
spark.sql("cache table df10_square"):
- 임시 뷰 df10_square를 캐시함.
spark.sql("uncache table df10_square"):
- 임시 뷰 df10_square를 캐시에서 제거함.
spark.sql("cache lazy table df10_square"):
- 임시 뷰 df10_square를 지연 캐시함.
spark.catalog.isCached("df10_square"):
- 임시 뷰 df10_square가 캐시되어 있는지 확인함.
spark.catalog.clearCache():
- 모든 캐시를 제거함.
spark.catalog.isCached("df10_square"):
- 임시 뷰 df10_square가 캐시되어 있는지 확인함.
spark.sql("cache table df10_square"):
- 임시 뷰 df10_square를 캐시함.

이 코드는 데이터 프레임의 생성과 캐싱, 그리고 캐시 제거 과정을 통해 Spark에서의 캐싱 메커니즘을 실험함으로써, Spark의 성능 최적화 기법을 설명함.

PERSIST

persist는 기본적으로 caching되는 데이터 프레임을 메모리와 디스크에 보관하고 복제도 수행함.

persist - 인자

persist는 인자를 통해 세부 제어 가능
- useDisk = True
- useMemory = True
- useOffHeap = Fasle
  - off Heap 설정이 필요
- deserialized = False
  - 메모리를 줄일지 아니면 cpu 계산을 줄일지
  - deserialized = True는 메모리에서만 가능
- replication = 1
  - 몇개의 복사본을 서로 다른 executor에 저장할지 결정

Dynamic-partition-pruning

Partition Pruning과 Execution Plan

Partition Pruning은 Logical Plan Optimization 단계에서 발생
개발자가 만든 spark sql query나 dataframe이 액션의 형태로 job으로 실행이 됨
- 이때 첫 번째로 하는 일은 주어진 쿼리를 분석함.
- 이후 logical plan으로 만들어낸 이후에 logical Optimization을 하는데 이때 다양한 기술이 쓰임(이중 partition Pruning이 있음)
- constant folding은 2+3을 컴파일 단계에서 계산하지 않고 미리 계산한다는거
- boolean expression simplification같은 경우에는 boolean을 복잡한 bool 계산을 단순화시킴

static Partition Pruning의 문제

Partitioning은 보통 큰 테이블에 적용되어 있음 (Fact table)

order 테이블에서 자주 조인되는 것이 아래에 있는 year_month 테이블이 있다고 가정

Fact table이 Dimension 테이블 조인 시 필터링이 Dimension 테이블에 적용되어 있다면?
만약 위에 저러한 sql 쿼리문이 있다고 가정.
orders와 date를 조인을 하는 과정의 그림
date 테이블 즉 디멘션 테이블 쪽에서 필터링이 된 상태에서 조인을 하면 굉장히 큰 낭비가 됨.(orders 테이블이 엄청 커서, fact table)
데이터를 줄인 상태에서 조인하는 것이 좋음

Dynamic Partition Pruning이란?

비 Partition 테이블에 적용된 필터링을 Partition 테이블에 적용해보는 것.
- 후자가 작은 차원의 테이블이라면 broadcast join까지하면 좋음

Dynamic Partition Pruning 예제

기본적으로 활성화 되어 있음.
- spark.sql.optimizer.dynamicPartitionPruning.enabled: true

기본적으로 이런 식으로 병렬 처리 됨.

Repartition-Coalesce

파티션의 수를 조절하는 방법
- Repartition (파티 션 수 늘릴 때)
- Coalesce(파티션 수를 줄일 때)

Repartition을 하는 이유

전체적으로 파티션의 수를 늘려 병렬성 증가를 위해(그러나 기본적으로 셔플링이 발생됨)
굉장히 큰 파티션이나 Skew 파티션의 크기를 조절하기 위해서
파티션의 분석 패턴에 맞게 재분배(Write once, read many)
- 어떤 DataFrame을 특정 컬럼 기준으로 그룹핑을 하거나 필터링을 자주 하는 경우
  - 미리 그 컬럼 기준으로 저장해두었다면 그게 Bucketing
Pepartition 방식
2가지 방식이 존재
- repartition
- repartitionByRange
Shuffling이 발생함. 분명한 이유를 가지고 Repartition을 사용해야 함.
- 많은 경우 repartition이 별 이유없이 사용되어 오히려 시간과 비용 증가.
  - 비슷하게 불필요하게 counting과 distinct counting과 duplicate제거가 비용 발생함.
Column이 사용되면 균등한 파티션 크기를 보자할 수 없음.
파티션의 수를 줄이는 용도로는 사용 불가
- 줄이는 경우에는 Coalesce

사용 방법

Hash 기반 Partitioning

repartition(5)
- 데이터 프레임을 5개의 파티션으로 나눔.
- 데이터 균등 분배가 필요할 때 사용함.
repartition(5, "city")
- city 컬럼을 기준으로 데이터 프레임을 5개의 파티션으로 나눔.
- city 컬럼 값에 따라 데이터 분배가 필요할 때 사용함.
repartition(5, "city", "zipcode")
- city와 zipcode 컬럼을 기준으로 데이터 프레임을 5개의 파티션으로 나눔.
- city와 zipcode 값을 조합하여 데이터 분배가 필요할 때 사용함.
repartition("city")
- city 컬럼을 기준으로 데이터 프레임을 파티셔닝함.
- city 값에 따라 데이터를 분배해야 할 때 사용함.
repartition("city", "zipcode")
- city와 zipcode 컬럼을 기준으로 데이터 프레임을 파티셔닝함.
- city와 zipcode 값을 조합하여 데이터를 분배해야 할 때 사용함.
repartitionByRange(numPartitions, *cols)
- 지정된 컬럼 값의 범위를 기준으로 파티션을 나누는 방식
- 데이터 샘플링 기반으로 파티션을 나누기에 결과가 매번 다를 수 있음.

테스트

이건 파티션을 10개로 나눈 것
라운드 로빈이라 각 파티션 크기가 균등

Coalesce가 필요한 경우

파티션 수를 줄이는 용도
셔플링이 발생시키지 않고 local partition을 merge
- 따라서 skew 파티션을 만들어낼 수 있음.
균등한 파티션 크기를 보장하지 않음.

Dataframe-hints

우리가 생각하는 방식보다 더 안좋은 최적화 방식이 채택될 수 있음.
즉 우리가 생각하고자 하는, 원하는 방향의 최적화로 하고자 하는 것
결론은 Spark SQL Optimizer에게 Execution plan을 만듬에 있어서 특정한 방식을 사용하도록 제안.
두 종류의 힌트가 존재
- Partitioning 관련 힌트
- join 관련 힌트

Dataframe - Partitioning 관련 힌트

❖ COALESCE

설명: 파티션 수를 줄이는 데 사용됨. 데이터 이동을 최소화하며 재분배함.
사용 예: df1.join(df2, "id", "inner").hint("COALESCE", 3)

❖ REPARTITION

설명: 파티션 수를 지정하여 데이터를 재분배함. 데이터 이동이 많이 발생할 수 있음.
사용 예: SELECT /*+ REPARTITION(3) */ * FROM TABLE

❖ REPARTITION_BY_RANGE

설명: 특정 컬럼의 값을 기준으로 데이터를 범위별로 재분배함.
사용 예: df.repartitionByRange(3, "col1")

❖ REBALANCE

설명: 파티션의 크기를 비슷하게 맞추기 위해 데이터를 재분배함. 주로 파일 저장 시 사용.
사용 예: df.rebalance("col1")

DataFrame Join 관련한 힌트들

❖ BROADCAST, BROADCASTJOIN, MAPJOIN

설명: 작은 테이블을 모든 노드에 브로드캐스트하여 조인 성능을 향상시킴.
사용 예: join_df = df1.join(df2.hint("broadcast"), "id", "inner")

❖ MERGE, SHUFFLE_MERGE, MERGEJOIN

설명: 기본 조인 전략으로, 두 테이블을 셔플하여 병합하는 방식.
사용 예: SELECT /*+ MERGE(df2) */ * FROM df1 JOIN df2 ON df1.order_month = df2.year_month

❖ SHUFFLE_HASH

설명: 해시 함수를 사용하여 데이터를 셔플하고 조인함. Full Outer Join에는 사용 불가.
사용 예: join_df = df1.join(df2.hint("SHUFFLE_HASH"), "id", "inner")

❖ SHUFFLE_REPLICATE_NL

설명: 크로스 조인을 수행하기 위해 모든 데이터를 셔플하고 복제하여 조인함.
사용 예: join_df = df1.crossJoin(df2).hint("SHUFFLE_REPLICATE_NL")

DataFrame 힌트 사용법

❖ SPARK SQL

형식: /*+ hint [, … ] */
예 1: SELECT /*+ REPARTITION(3) */ * FROM TABLE
예 2: SELECT /*+ BROADCAST(table1) */ * FROM table1 JOIN table2 ON table1.key = table2.key

❖ DataFrame API

형식: .hint 메소드 사용
예 1: join_df = df1.join(df2, "id", "inner").hint("COALESCE", 3)
예 2: join_df = df1.join(df2.hint("broadcast"), "id", "inner").hint("COALESCE", 3)

Spark - AQE(Adaptive Query Execution)

AQE 이전(spark 3.1 이전)

spark 1.x는 Catalyst Optimizer 와 Tungsten Project
- 전자는 규칙 기반 최적화 수행
- 후자는 기본적으로 jvm 문제 없이 코드 최적화를 하려는 것(GC를 피하기 위해 직접 off heap 메모리 관리 수행)
spark 2.x CBO(Cost-based-Optimizer)
- 데이터프레임 통계정보 이용해 효율적인 execution plan 생성
  - 전체 크기, 레코드 수, 컬럼별 특성을 보고 비용을 계산하게 됨.

만약 이러한 테이블을 기준으로 다음과 같은 쿼리를 한다고 함.

그럼 group by 쿼리는 2개의 stage를 만들어냄
spark.sql.shuffle.partitions 값에 의해 Shuffling 후 Partition 수 결정.
spark.sql.shuffle.partitions
- 이 변수 하나로 다양한 상황의 shuffling을 해결하기는 쉽지 않음.
- 적은 수의 partition은 병렬성을 낮추고 OOM과 disk spill이 가능성을 높임.
- 많은 수의 partition은 task scheduler와 task 생성과 관련된 오버헤드가 생기며 너무 흔한 네트워크 I/O 요청으로 병목 초래
이러한 의문이 발생함 " 만약에 spark engine Optimizer 가 알아서 partition의 수를 결정할 수 있다면?"

이 이슈를 해결하기 위한 것

대용량 데이터 베이스 분야에서는 이 문제는 잘 연구된 문제.
기본적인 아이디어는 parsing time 최적화와 run time 최적화의 병행
- Pasing time 정보로 선택된 physical plan과 코드 최적화만으로는 불충분
- 특히 UDF가 많이 사용되는 경우 이 문제는 더 심각해짐

AQE

실행 시간 통계 정보를 기반으로 쿼리 실행 중간에 동적으로 최적화 하는 방법.
- 원리는 AQE는 모든 최적화 결정을 정확한 실행 시간 통계에 기반하여 수행
최적의 시점.
- 쿼리가 query -> job -> stage -> task 순으로 진행될 때, 최적화 방식에 변경을 줄 수 있는 최적의 시점은 stage가 완료된 후임.
- 이는 실행 시간 통계 정보를 가장 잘 반영할 수 있는 시점이기 때문.

Stage가 가장 좋은 최적화 방식 변경 포인트

결국 데이터 전송이 이뤄지는 셔플링과 브로드캐스팅은 스테이지가 변할때 즉 스테이지간의 변화에서 발생함.
결국 이 시점에 많은 경우 데이터들이 materialize가 됨. (통계정보를 알 수 있는 시점)

AQE 이후

앞서 group by 쿼리 2번째 스테이지 시작부에 AQEShuffleRead가 사용
저기서 무슨 일을 하냐면, 셔플링 데이터 크기를 보고 동적으로 파티션의 수를 조정하는 것.

AQE가 필요한 경우

셔플링 후에 파티션 수를 동적으로 조정하는 경우(기본적으로 줄여나가는 식으로 감)(spark 3)
조인 방법을 바꾼다. (spark 3.2)
skew된 join을 할때 최적화를 하는 것(spark 3)

Dyinamic coalescing (post) shuffle Partitions

stage DAG를 순차적 실행
매번 새로운 최적화 기회가 있는지 조사
- 필요하다면 다시 실핼하거나 쿼리 플랜 변경
사진과 같이 1과 2를 조인하고 3을 그룹바이하고 4를 오퍼레이션이라고 가정할 수 있음.
이러한 plan은 스태틱하게 한 번 만들어지고 끝이나, AQE가 inable가 된다면 다음 사진과 같이 됨
stage 1을 실행해보고 최적화할 기회가 있는지 계속 옅보는 것.

이게 무엇인가?

먼저 왜 필요한가?
- 적당한 파티션의 크기와 수는 성능에 지대한 영향을 끼침
- 너무 많은 수의 작은 파티션
  - 스케줄러 오버헤드
  - 태스크 준비 오버 헤드
  - 비효율적인 I/O
- 적은 수의 큰 파티션
  - OOM
  - disk Spill
  -Spark.sql.shuffle.partitions라는 하나의 변수로는 불충분
동작 방식

scan t
- table t를 스캔하여 데이터 읽어옴
Aggregate
- 읽어온 데이터를 그룹화하여 v의 합계 계산
shuffle
- 데이터를 셔플하여 그룹화된 데이터를 재분배
aggregate
- 셔플된 데이터를 다시 그룹화하여 최종 합계 계산
shuffle
- 데이터를 다시 셔플하여 정렬 준비를 함.
sort
- 데이터를 v의 합계 기준으로 내림차순 정렬

COALESCE
- 첫 번째 셔플 이후에 COALESCE 연산을 추가하여 파티션 수를 다시 줄임.
- 이후 단계를 통해 파티션 수가 줄어든 데이터를 처리함.

COALESCE
- 두 번째 셔플 이후에 COALESCE 연산을 추가하여 파티션 수를 다시 줄임.
- 최종 정렬 단계까지 파티션 수가 데이터를 처리함.

환경 변수 이름	기본 값	설명
spark.sql.adaptive.coalescePartitions.enabled	True	spark.sql.adaptive.enabled도 true인 경우 셔플 후 파티션 수를 동적으로 줄이며 파티션의 크기는 아래 변수(advisoryPartitionSizeInBytes)로 맞추려 시도
spark.sql.adaptive.advisoryPartitionSizeInBytes	64MB	셔플링 후 파티션 수를 줄일 때 목표로 하는 파티션의 크기
spark.sql.adaptive.coalescePartitions.parallelismFirst	True	이 값이 true이면 병렬성 보장을 위해 위의 목표 크기가 무시되고 아래 minPartitionSize만 보장
spark.sql.adaptive.coalescePartitions.initialPartitionNum	없음	Coalescing 전의 파티션 수. 없으면 spark.sql.shuffle.partitions로 설정
spark.sql.adaptive.coalescePartitions.minPartitionSize	1MB	Coalescing 후 파티션의 최소 크기

Dynamically Coalescing Shuffle Partitions 동작 방식

AQE의 해법

내부적으로 많은 수의 파티션을 일부러 생성
- 설정 변수: spark.sql.adaptive.coalescePartitions.initialPartitionNum (기본값: 200)
- 초기 단계에서 많은 수의 파티션을 생성하여 작업 병렬성을 극대화함.
매 Stage가 종료될 때 필요하다면 자동으로 Coalesce 수행
- 설정 변수: spark.sql.adaptive.coalescePartitions.enabled
- 각 스테이지가 끝날 때 실행 시간 통계를 바탕으로 파티션 수를 동적으로 줄임.
설정에 따라 파티션의 크기는 최소 크기 혹은 목표 크기를 맞추려 동작
- 설정 변수:
  - spark.sql.adaptive.advisoryPartitionSizeInBytes: 셔플링 후 파티션 수를 줄일 때 목표로 하는 파티션의 크기 (기본값: 64MB)
  - spark.sql.adaptive.coalescePartitions.minPartitionSize: Coalescing 후 파티션의 최소 크기 (기본값: 1MB)
  - spark.sql.adaptive.coalescePartitions.parallelismFirst: 병렬성 보장을 위해 목표 크기 무시 여부 결정 (기본값: True)
- 파티션의 크기는 최소 크기 또는 목표 크기로 조정되며, 병렬성 보장을 위해 설정값에 따라 조정 방식을 결정함.

AQE 데모

2개의 table을 만들어서 3가지 aqe 기능을 테스트하는데 사용
테이블 1번 (items): 일종의 dimension 테이블
- 30,000,000개의 레코드가 존재
- id와 price 필드로 구성
테이블 2번 (sales) : 일종의 fact 테이블
- 1,000,000,000개의 레코드 존재
- item_id와 quantity와 date 필드로 구성
위 두 개의 테이블을 spark sql을 사용해서 생성

이영준

개인공부용(업데이트 중단)

이전 포스트

spark 고급,spark ML - 1

다음 포스트