Amazon EMR의 Apache Spark 작업에서 “디바이스에 남은 공간 없음” 스테이지 실패를 해결하려면 어떻게 해야 합니까?

간략한 설명

다음 이유 중 하나로 인해 Apache Spark 애플리케이션에서 디바이스에 남은 공간 없음 오류가 발생할 수 있습니다.

• 셔플 조인으로 인해 셔플 프로세스 중에 상당량의 중간 데이터 생성
• 고르지 않은 데이터 파티션의 분포 및 실행기 워크로드 분포
• 부적합한 파티션 크기 및 수
• 충분하지 않은 리소스(예: 디스크 및 메모리) 가용성

Apache Spark는 코어 및 태스크 노드의 로컬 스토리지를 사용하여 중간(셔플) 데이터를 저장합니다. 인스턴스의 디스크 공간이 부족하면 작업이 실패하고 디바이스에 남은 공간 없음 오류가 발생합니다.

해결 방법

이 문서에서는 디바이스에 남은 공간 없음 오류의 가장 빈번한 원인과 해결 방법을 설명합니다. 적절하게 수정하려면 근본 원인을 식별해야 합니다.

참고: AWS Command Line Interface(AWS CLI) 명령을 실행할 때 오류가 발생하면 AWS CLI의 오류 해결을 참조하십시오. 또한 최신 AWS CLI 버전을 사용하고 있는지 확인하십시오.

재파티셔닝

클러스터의 코어 및 태스크 노드 수에 따라 Spark 파티션 수를 늘려야 할 수 있습니다. Spark 파티션을 추가하려면 다음 명령을 실행합니다.

val numPartitions = 500
val newDF = df.repartition(numPartitions)

참고: 500을 사용 사례에 맞는 파티션 수로 바꾸십시오.

Spark 구성 조정

파티션 관리

셔플 중에 디스크 스필이 너무 많거나 파티션 간에 데이터가 고르지 않게 분배되면 다음 파라미터를 조정하십시오.

spark.default.parallelism=${NUM_CORES * 2} #no. of partitions in RDDs
spark.sql.shuffle.partitions=${NUM_CORES * 2} #no. of shuffle partitions
spark.sql.files.maxPartitionBytes=256MB #max. no. of bytes in a partition when reading files
spark.sql.files.maxRecordsPerFile=10000000

다음 중 하나에 해당하는 경우 병렬 처리 수와 파티션 수를 늘리십시오.

• 태스크 지속 시간 편차가 평균 지속 시간보다 3배 이상 큼
• 태스크당 스필이 20%를 초과함

평균 파티션 크기가 50MB 미만이거나 작은 파일이 너무 많으면 병렬 처리 수와 파티션 수를 줄이십시오.

최적의 파티션 수를 계산하려면 다음 공식을 사용합니다.

Initial Partitions = Number of Cores * 2
Optimal Partitions = max(
    Total Input Size / Target Partition Size,
    Initial Partitions
)

데이터 볼륨 기반 조정

다음은 다양한 크기의 데이터세트에 대한 구성 파라미터입니다.

소규모 데이터세트(100GB 미만):

spark.sql.files.maxPartitionBytes=128MB
spark.sql.shuffle.partitions=NUM_CORES * 2
spark.sql.files.maxRecordsPerFile=5000000

중규모 데이터세트(100GB~1TB):

spark.sql.files.maxPartitionBytes=256MB
spark.sql.shuffle.partitions=NUM_CORES * 3
spark.sql.files.maxRecordsPerFile=10000000

대규모 데이터세트(1TB 이상):

spark.sql.files.maxPartitionBytes=512MB
spark.sql.shuffle.partitions=NUM_CORES * 4
spark.sql.files.maxRecordsPerFile=20000000

메모리 및 스토리지 최적화

메모리와 스토리지를 최적화하려면 구성 파라미터를 업데이트하십시오.

spark.memory.fraction=0.8 # Higher for compute-intensive jobs
spark.memory.storageFraction=0.3 # Lower for shuffle-heavy workloads
spark.executor.memoryOverhead=0.2 # 20% of executor memory
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=${EXECUTOR_MEMORY * 0.2}

Spark 실행기 컨테이너의 총 메모리 할당량을 계산하려면 다음 네 가지 메모리 구성 요소를 결합합니다.

• 실행기 메모리(spark.executor.memory)
• 메모리 오버헤드(spark.executor.memoryOverhead)
• Off-Heap 메모리(spark.memory.offHeap.size)
• PySpark 메모리(spark.executor.pyspark.memory)

총 실행기 컨테이너 메모리 = spark.executor.memory + spark.executor.memoryOverhead + spark.memory.offHeap.size + spark.executor.pyspark.memory

다음 계산에 따라 Spark의 내부 메모리 할당이 결정됩니다.

Storage Memory = Executor Memory * memory.fraction * memory.storageFraction
Execution Memory = Executor Memory * memory.fraction * (1 - memory.storageFraction)
Off-Heap Memory = Executor Memory * 0.2

파일 및 디스크 관리

파티션 편차가 있는 경우, 또는 파티션 수가 너무 많거나 적은 경우 파일 관리 구성을 조정하십시오. 사용 사례에 따라 다른 값이 필요한 경우를 제외하면 maxPartitionNum을 총 코어 수의 2배로 설정하고 minPartitionNum을 1로 설정하십시오.

# File Management
spark.sql.files.minPartitionNum=${NUM_CORES}
spark.sql.files.maxPartitionNum=${NUM_CORES * 4}
spark.shuffle.file.buffer=64k

maxPartitionNum을 너무 작게 설정하면 병렬 처리가 제한되고 전체 편차 시나리오를 방지하지 못할 수 있습니다.

AQE 및 편차 처리

Spark의 적응형 쿼리 실행(AQE)은 실시간 통계를 기반으로 쿼리 계획을 조정하는 런타임 최적화입니다.

AQE는 Amazon EMR 버전 5.30.0 이상에서는 기본적으로 켜져 있습니다. Spark의 AQE는 조인 전략과 셔플링을 자동으로 최적화할 수 있습니다. 또한 동적 파티션 분할을 통해 데이터 편차를 효과적으로 처리할 수 있습니다. 이를 통해 로드 밸런싱과 쿼리 성능이 개선됩니다.

# Skew Management
spark.sql.adaptive.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.skewJoin.skewedPartitionFactor=10
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB

AQE를 지원하지 않는 이전 버전의 Spark를 사용하는 경우 다음 방법 중 하나를 사용하여 데이터 편차를 관리하십시오.

• spark.sql.autoBroadcastJoinThreshold 임계값에서 브로드캐스트 조인을 조정합니다. 이는 조인에서 한 데이터세트가 다른 데이터세트보다 훨씬 작을 때 유용합니다.
• 코드에서 repartition() 또는 **coalesce()**를 사용하여 데이터 분배를 개선합니다.
• 더 크거나 편향된 테이블에 SKEW 힌트를 적용하고 작은 테이블을 브로드캐스트합니다. SKEW 힌트는 테이블에 편향된 데이터가 있음을 Spark 최적화 프로그램에 알리고 조인 전략 최적화에 도움을 줍니다.

다음은 Spark SQL 쿼리의 SKEW 힌트 예시입니다.

-- Using SKEW hint in Spark SQL
SELECT /*+ SKEW('t1') */
    t1.key, t2.value
FROM table1 t1 JOIN table2 t2 ON t1.key = t2.key

-- Using MAPJOIN hint along with SKEW
SELECT /*+ SKEW('t1'), MAPJOIN(t2) */
    t1.key, t2.value
FROM table1 t1
JOIN table2 t2
ON t1.key = t2.key

스토리지 동적 스케일 업을 위한 부트스트랩 작업

Amazon CloudWatch 모니터링 및 Lambda 자동화를 통한 부트스트랩 작업을 사용하여 Amazon EMR 클러스터의 스토리지를 자동으로 스케일링할 수 있습니다. 사용 가능한 디스크 공간이 설정된 임계값 아래로 떨어지면 Amazon CloudWatch는 AWS Lambda 함수를 시작합니다. 이 함수는 새 Amazon Elastic Block Store(Amazon EBS) 볼륨을 클러스터 노드에 연결합니다. 그런 다음 이 함수는 스크립트를 실행하여 볼륨을 포맷, 마운트하고 Hadoop 분산 파일 시스템(HDFS)에 통합합니다.

자동화된 접근 방식으로 스토리지 제약으로 인한 클러스터 장애를 방지하고 필요할 때만 용량을 추가함으로써 비용 효율성을 유지합니다. 구현에는 적절한 Identity and Access Management(IAM) 역할, Amazon CloudWatch 경보, AWS Lambda 구성, 볼륨 관리를 위한 사용자 지정 스크립트가 필요합니다. 자세한 내용은 Amazon EMR 클러스터의 동적 스토리지 스케일 업을 참조하십시오.

Amazon EBS 용량 추가

새 클러스터의 경우 더 큰 EBS 볼륨 사용

Amazon EMR 클러스터를 시작하고 EBS 볼륨이 더 큰 Amazon Elastic Compute Cloud(Amazon EC2) 인스턴스 유형을 선택합니다. 자세한 내용은 인스턴스의 기본 Amazon EBS 스토리지를 참조하십시오.

실행 중인 클러스터의 경우 EBS 볼륨 추가

다음 단계를 완료합니다.

1. 더 큰 EBS 볼륨으로도 문제가 해결되지 않으면 코어 및 태스크 노드에 추가로 EBS 볼륨을 연결합니다.

2. 연결된 볼륨을 포맷 및 마운트합니다. 올바른 디스크 번호(예: /data 대신 /mnt1 또는 /mnt2)를 사용해야 합니다.

3. SSH 클라이언트를 사용하여 노드에 연결합니다.

4. /mnt2/yarn 디렉터리를 생성하고, 디렉터리 소유권을 YARN 사용자에게 설정합니다.

   sudo mkdir /mnt2/yarn
   sudo chown yarn:yarn /mnt2/yarn

5. /mnt2/yarn 디렉터리를 /etc/hadoop/conf/yarn-site.xml의 yarn.nodemanager.local-dirs 속성에 추가합니다.
   예시:

   <property>
       <name>yarn.nodemanager.local-dirs</name>
       <value>/mnt/yarn,/mnt1/yarn,/mnt2/yarn</value>
   </property>

6. NodeManager 서비스 재시작:
   Amazon EMR 4.x-5.29.0 릴리스 버전

   sudo stop hadoop-yarn-nodemanager
   sudo start hadoop-yarn-nodemanager

   Amazon EMR 5.30.0 이상 릴리스 버전

   sudo systemctl stop hadoop-yarn-nodemanager
   sudo systemctl start hadoop-yarn-nodemanager

관련 정보

Amazon EMR에서 Spark 작업의 스테이지 실패 문제를 해결하려면 어떻게 해야 합니까?

EKS의 Amazon EMR이란 무엇입니까?

Amazon EMR Serverless란 무엇입니까?

AWS Open Data Analytics 웹사이트의 모범 사례