[Spark] TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again. 오류 발생 시 해결 방법

뮨제상황

dataframe 두 개를 조인하려는데 아래와 같은 워닝이 주르륵 발생하더니 오류를 출력하고 어플리케이션이 종료됐다.

...
23/05/22 04:39:35 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:35 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:35 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:36 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:36 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:36 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:36 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:37 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:37 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:38 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:38 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:38 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:39 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:39 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:39 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:39 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:40 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:40 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:40 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:41 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:41 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:41 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
23/05/22 04:39:41 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again.
...

 

이 에러는 TaskMemoryManager가 페이지를 할당할 때 발생하는 것 같다. 

(참고:

 https://github.com/apache/spark/blob/d3f12df6e09ee47dbd7c9e08c9962430a2601941/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java#L334)

resourceManager에게 이전에 메모리 할당을 허락받은 후에 `tungstenMemoryAllocator`를 이용해 메모리를 할당받고 `pageNumber`를 업데이트 한다. 이때 OutOfMemory가 발생하면 해당 에러를 발생시킨다.

 

문제의 상황에서는 한 쪽의 데이터프레임 사이즈가 작았다. (skewed data)

실제로  dag를 살펴보니 broadcastHashJoin을 사용하고 있었다.

데이터사이즈가 충분히 작은 것 같은데 왜 여기서 오류가 나는지는 모르겠다 ㅠ

 

 

해결방법1. 작은 쪽의 데이터프레임 크기를 임의로 키우기

small_data = small_data.union(small_data).union(small_data).union(small_data).union(small_data).union(small_data).union(small_data).union(small_data)
big_data.join(small_data.distinct(), [conditions], how='left')

야매긴 한데 작은 데이터를 union을 통해 크기를 키워준 뒤, join 시 distinct()를 통해 중복제거하고 사용하면 오류 없이 정상적으로 실행됐다. 

이 경우에는 broadcast hash join을 사용하지 않았다.

 

해결방법2. executor 메모리 크기 늘리기

spark.executor.memory

 

해결방법3. broadcastjoin 비활성화

아래의 config 값을 -1로 설정하면 비활성화되어 다른 조인 방법을 사용했다.

spark.sql.autoBroadcastJoinThreshold

그런데 이상한건 이게 될 때도 있고 안될때도 있어서 조건은 확인해봐야할 것 같다.

 

 

추가: Project Tungsten

TaskMemoryManager.java에서 볼 수 있다시피 메모리 관련 함수나 변수명에 `tungsten`이라는 단어가 많이 등장해서 궁금해서 찾아봤다.

Tungsten이라는 프로젝트가 있는데 Spark 2부터 도입된 엔진이다.

디스크I/O나 네트워크 쪽을 개선하면서 성능 향상을 도모하는 다른 프로젝트와는 달리 스파크의 CPU와 메모리 관리에 기여하고 이 부분에서 많은 성능 향상을 책임지고 있다고 한다.

(sun.misc.Unsafe 사용) Tungsten은 JVM의 heap memory를 사용하지 않고 native 영역 메모리를 사용하여 이러한 문제를 해결했다고 한다. 

참고로 sparkSQL 사용 시 최적화에도 사용한다: Catalyst optimizer로 논리적 쿼리 계획에서 물리적 쿼리 계획을 생성한 후에 Tungsten의 Codegen 기능을 사용하여 최적화된 코드를 생성한다고 한다.

 

Tungsten의 역할은 크게 아래와 같다.

1. Off-Heap Memory Management: 메모리를 관리

2. Cache Locality: 캐시를 잘 배치..(for high cache hit rate)

3. CodeGen: Whole-Stage Code Generation

 

관련 속성으로는 `spark.sql.tungsten.enabled` 가 있다.

 

(참고: https://www.linkedin.com/pulse/catalyst-tungsten-apache-sparks-speeding-engine-deepak-rajak)