How LocalShuffleReaderExec works

Task는 파티션 단위로 이루어진다.

파티션이 100개면 Task는 최소 100개 이상일 것. Worker가 일을 하는 동안 Task가 어디로 부여되면 좋을지 판단하는 방법은 Data locality?

데이터에 100개의 row가 있을 때, worker node 3개에 각각에 10, 70, 10개의 row를 가지고 있는 worker 중에서는 70개를 가진 node에 들어가야 30개만 network로 가져와야 할 것..

Spark Scheduling Layer

어느 Node에 가서 일하는 것이 좋을까? TaskSetManager의 역할. AQE가 어떤 데이터가 어떤 노드에 얼마나 있는지를 계산해서 어떤 Node에 가야 할 지 우선순위를 정해준다. 3→7→4→… 이러면 노드 3을 확인하고, 바쁘면 7에도 요청하는 식이다.

Locality Levels

• PROCESS_LOCAL → 같은 Executer 안에 있어서 가장 높은 locality
• NODE_LOCAL → 같은 노드에 있지만, 다른 Executor(JVM)에 있을 때, 같은 node에 있는 LocalShuffleReaderExec가 shuffle을 해준다.
• NO_PREF → 아무거나 괜찮다. 왜냐하면 클라우드 환경에서는 RACK이 의미가 없기 때문에. + Preference Node에 대한 정보는 없지만 locality를 찾으려고 최대한 노력했다.
• RACK_LOCAL → 같은 랙에 있는 것
• ANY → 같은 랙이라는 정보도 없는 최악의 상황

아래로 level이 떨어지는 것을 relax라고 함.

우리가 처음부터 ANY를 지정하는 것이 아니라, TaskManager가 위에서부터 차례대로 시도하는 것인데, ANY가 된다는 것은 최악의 상황이 된 것이다.

Trade-off

더 오래 기다리면 locality가 증가해서 remote fetch가 감소하겠지만, 그만큼 놀고 있는 CPU가 더 많아진다.

반대로 짧게 기다리면 CPU의 처리량이 증가하지만 network 사용량이 많아진다. 오히려 remote fetch를 하는 동안 CPU가 놀게 될 수도 있다.

Locality는 지키면 좋지만, 반드시 보장되는 것은 아니다.

Problem - LocalShuffleReaderExec 이전에는

생각 없이 네트워크로부터 마구 fetch해왔다.