zeppelin으로 spark 사용해 Elasticsearch 업데이트를 수행 할 때 가장 어려운 부분이 색인 작업 시간이다.
Spark 튜닝은 전문작업자가 하기에도 어려운 작업들이 많고 ES 튜닝도 쉽지않다.
다른 건 몰라도 하기 2개의 튜닝 포인트는 꼭 잡으면 색인에 소요되는 시간을 어느정도 줄일 수 있다.
1. Spark partition 튜닝
spark partition 은 다음과 같이 3가지 종류가 있다.
- Inpout, output, shuffle
- input은 말그대로 DB나 파일을 읽어 올 때 사용된다.(default: 134217728(128MB))
print(spark.conf.get("spark.sql.files.maxPartitionBytes")) - output은 이를 다시 DB나 파일로 저장할 때 사용된다.
spark.conf.set("spark.sql.files.maxRecordsPerFile",num of records)
write시에 .option("maxRecordsPerFile",100000) 으로도 지정 가능 - shuffle은 연산 중 join, groupby와 같은 작업에 수행된다.(default: 200)
print(spark.conf.get("spark.sql.shuffle.partitions"))
읽어 들어야 할 파일의 크기가 크고 join, groupBy와 같은 작업량이 많이 요구 된다면 총 파일크기 기준으로 파티션 값을(shuffle.partitions) 100~200mb 사이로 맞추는 작업이 중요하다.
250GB(DB or 파일) / 140MB = 약 1800 개
500GB(DB or 파일) / 140MB = 약 3600 개
예) 사용하는 DB가 250GB라면 shuffle.partitions 의 값은 1800개로 셋팅 후 작업한다.
spark.conf.set("spark.sql.shuffle.partitions", 1800 )
※ 이 외 sql 상세 튜닝은 여기 에서 확인하면 된다.
이 외에도 zeppelin에서 수행되는 spark conf 상세 튜닝이 가능한다.
하기 이미지와 같이 executor를 동적으로 할당하는 설정을 튜닝 할 수 있다.
"spark.dynamicAllocation.initialExecutors": Dynamic Allocation application이 시작될 때 executor 수
"spark.dynamicAllocation.maxExecutors": Dynamic Allocation 최소한의 executor 수
"spark.dynamicAllocation.minExecutors": Dynamic Allocation 최대한의 executor 수
※ 이 외 spark configuration 상세 튜닝은 여기에서 확인!
2. Spark to ES bulk 옵션 튜닝
Spark에서 아무리 파티셔닝과 MR을 잘 처리 했더라도 결국 ES업데이트 관련 업데이트 설정을 최적화해 주어야 한다.
spark에서 batch job이 돌면서 ES의 bulk API를 사용한다. 이에 batch size를 얼마나 둘지 등을 지정한다.
관련 설정은 하기와 같다.
es.batch.size.bytes(default 1mb)
es.batch.size.entries(default 1000)
es.batch.write.refresh (default true)
es.batch.write.retry.count (default 3)
es.batch.write.retry.wait (default 10s)
상세 설정은 여기
참조
spark partition 에 대해 잘 설명해둔 블로그:
https://spidyweb.tistory.com/312
https://tech.kakao.com/2021/10/08/spark-shuffle-partition/
spark configuration 튜닝:
https://spidyweb.tistory.com/331