Spark SQL 작은 파일 문제 OPPO 솔 루 션

본 고 는 OPPO 인터넷 기초 기술 팀 에서 온 것 으로 주석 작성 자 를 전재 합 니 다.동시에 저희 공식 번호 에 관심 을 가 져 주 셔 서 감사합니다. OPPOtech, OPPO 최첨단 인터넷 기술 과 활동 을 공유 합 니 다.
Spark SQL 작은 파일 은 hdfs block 블록 크기 보다 파일 크기 가 현저히 작은 파일 을 말 합 니 다.너무 많은 작은 파일 은 HDFS 에 심각 한 성능 병목 을 가 져 오고 작업 의 안정 과 클 러 스 터 유지 에 큰 도전 을 가 져 올 수 있 습 니 다.
일반적으로 Hive 를 통 해 배 치 된 MR 작업 은 다음 과 같은 작은 파일 을 합 친 매개 변 수 를 간단하게 설정 하여 작업 에 발생 하 는 작은 파일 문 제 를 해결 할 수 있 습 니 다.

set hive.merge.mapfiles=true;
set hive.merge.mapredfiles=true;
set hive.merge.size.per.task=xxxx;
set hive.merge.smallfiles.avgsize=xxx;

그러나 우리 가 오프라인 스케줄 링 임 무 를 Hive 에서 Spark 로 점차적으로 옮 기 는 과정 에서 Spark 자체 가 작은 파일 합병 기능 을 지원 하지 않 기 때문에 작은 파일 문제 가 날로 심각 해 지고 클 러 스 터 안정성 에 큰 영향 을 주 며 우리 의 이전 작업 을 방해 했다.
작은 파일 문 제 를 해결 하기 위해 저 희 는 처음에 파 라 메 터 를 계속 조정 하고 후기 코드 개발 등 다양한 단 계 를 거 쳐 간단 한 공 유 를 해 드 렸 습 니 다.
1. 스파크 가 왜 작은 파일 을 만 들 었 는 지
Spark 에서 생 성 된 파일 의 수량 은 RDD 에서 partition 의 수량 과 표 파 티 션 의 수량 에 달 려 있 습 니 다.여기 서 두 개의 파 티 션 개념 이 다 릅 니 다. RDD 의 파 티 션 은 작업 병행 도와 관련 되 고 표 파 티 션 은 Hive 의 파 티 션 수 입 니 다.생 성 된 파일 수 는 일반적으로 RDD 파 티 션 수 와 표 파 티 션 의 곱 입 니 다.따라서 작업 병행 도가 높 거나 파 티 션 수가 많 을 때 작은 파일 이 많이 생 성 된다.
따라서 생 성 된 파일 수 를 매개 변수 조정 으로 줄 여야 한다 면 마지막 단계 RDD 의 파 티 션 수 를 줄 여 도달 할 수 밖 에 없다 (파 티 션 수 를 줄 여 역사 데이터 와 상하 관계 로 제한 하고 수정 하기 어렵다).
2. 지역 사회 버 전의 매개 변 수 를 바탕 으로 조정 하 는 방안
2.1 Shuffle 연산 자가 없 는 간단 한 정적 파 티 션 SQL
이러한 SQL 은 비교적 간단 합 니 다. 주로 filter 상류 표 의 일부 데 이 터 를 하류 표 에 기록 하거나 두 장의 표 가 간단 한 UNION 에서 시작 하 는 작업 입 니 다. 이러한 작업 의 파 티 션 수 는 주로 파일 을 읽 을 때 Partition 수 에 의 해 결 정 됩 니 다.

Spark 2.4 이후 Hive orc 표 와 parquet 에 대한 지원 이 좋 았 기 때문에 운행 속 도 를 높이 기 위해 Hive orc / parquet 표를 자동 으로 DataSource 로 전환 하 는 매개 변 수 를 열 었 습 니 다.이러한 DataSource 표 의 유형 에 대해 partition 수 는 주로 다음 과 같은 세 가지 매개 변수 로 관 계 를 제어 합 니 다.

spark.sql.files.maxPartitionBytes；
spark.sql.files.opencostinbytes；
spark.default.parallelism；

그 관 계 는 다음 그림 과 같 기 때문에 이 세 개의 매개 변 수 를 조정 하여 데이터 의 블록 을 입력 하여 조정 할 수 있다.

DataSource 표 가 아 닌 CombineInputFormat 을 사용 하여 데 이 터 를 읽 기 때문에 주로 MR 매개 변 수 를 통 해 분 편 조정 을 합 니 다: mapreduce. input. fileinputformat. split. minsize

비록 우 리 는 입력 데이터 의 필름 을 조정 함으로써 최종 파일 의 수량 을 조정 할 수 있 지만 이러한 조정 은 불안정 하고 상류 데이터 의 크기 에 약간의 변화 가 발생 하면 파라미터 의 재배 치 를 가 져 올 수 있다.
이 문 제 를 간단하게 거 칠 게 해결 하기 위해 우 리 는 이러한 SQL 에 repartition 의 hint 를 추가 하여 새로운 shuffle 을 도입 하여 파일 의 수량 이 고정 값 임 을 보증 합 니 다.
2.2 Shuffle 연산 자가 있 는 정적 파 티 션 작업
ISSUE SPARK - 9858 에 새로운 인자 가 도입 되 었 습 니 다: spark. sql. adaptive. shuffle. targetPost Shuffle InputSize,
후기 에 spark adaptive 를 바탕 으로 이 매개 변 수 를 한층 더 강화 하여 파 티 션 의 수량 을 동적 으로 조정 할 수 있 습 니 다. 가능 한 한 모든 task 처리 targetPostShuffle Input Size 의 데 이 터 를 확보 할 수 있 기 때문에 이 매개 변 수 는 생 성 된 파일 의 수량 을 어느 정도 제어 할 수 있 습 니 다.
2.3 동적 분할 작업
동적 파 티 션 작업 은 파 티 션 이라는 변수 가 존재 하기 때문에 rdd 쪽 의 파 티 션 수 를 단순 하 게 조정 하면 전체 파일 수 를 제어 하기 어렵 습 니 다.
hive 에서 저 희 는 hive. optimize. sort. dynamic. partition 을 설정 하여 동적 파 티 션 에 파일 이 너무 많아 서 작업 을 수행 할 때 task 노드 가 자주 oom 하 는 상황 을 완화 할 수 있 습 니 다.이러한 매개 변 수 는 새로운 shuffle 을 도입 하여 데 이 터 를 재 정렬 하고 같은 partition 을 같은 task 에 나 누 어 처리 하여 하나의 task 가 여러 개의 파일 핸들 을 동시에 가지 고 있 는 것 을 피 할 수 있 습 니 다.
따라서 우 리 는 이러한 사상 을 빌려 distribute by 문 구 를 사용 하여 sql 을 수정 하여 파일 의 수량 을 제어 할 수 있다.일반적으로 각 파 티 션 에 N 개 이상 의 파일 을 만 들 지 않 으 려 면 SQL 말미 에 DISTRIBUTE BY [동적 파 티 션 열], ceil (rand () * N) 을 추가 할 수 있 습 니 다.
3. 파일 을 통합 할 수 있 는 commitProtocol 방안 을 자체 연구 합 니 다.
여러 가 지 를 종합해 보면 모든 방법 에 어느 정도 폐단 이 존재 하고 많은 규칙 도 실제 사용 과정 에서 업무 측 에 큰 어려움 을 준다.
그래서 우 리 는 spark 쪽 에서 hive 와 유사 한 작은 파일 통합 체 제 를 실현 하고 싶 습 니 다.몇 가지 가능 한 방안 선택 에서 우 리 는 최종 적 으로 spark. sql. sources. comitProtocolclass 를 다시 쓰 는 방법 을 선택 했다.
한편, 이 방안 은 Spark 코드 에 침입 하지 않 아 Spark 소스 코드 의 유지 에 편리 하 다. 다른 한편, 이 방안 은 업무 자 에 게 우호 적 이 고 set 명령 을 통 해 동적 으로 설정 할 수 있 으 며 문제 가 발생 하면 스크롤 백 하 는 것 도 매우 편리 하 다.업무 측 이 사용 하 는 과정 에서 spark. sql. sources. comitProtocolclass 를 간단하게 설정 하면 작은 파일 합병 여 부 를 제어 할 수 있 습 니 다.
작은 파일 통합 파 라 메 터 를 열 면 commt 단계 에서 생 성 된 모든 파일 을 가 져 와 새로운 job 를 도입 하여 이 파일 들 을 처리 합 니 다.우선 첫 번 째 job 에서 spark. copact. smallfile. size 보다 작은 모든 크기 의 파일 을 가 져 왔 습 니 다. 검색 이 끝 난 후에 spark. copact. size 매개 변수 값 에 따라 조합 파일 을 가 져 오고 두 번 째 job 에서 이 파일 들 을 합 쳤 습 니 다.