[Spark] Speculative Execution

분산 환경에서는 고르지 못한 리소스, 데이터 로드 등 task 실행 속도가 느려지는 상황이 많이 있다. 

Speculative Execution

한 단계에서 여러 작업의 실행 시간이 다를 수 있다. 다른 task와 비교하여 특히 느리게 실행되는 Task가 있을 수 있는데, 클러스터 내 노드의 구성 성능 차이, 네트워크 변동 등의 이유가 있을 수 있다.

speculative execution은 동일한 단계에 오랫동안 완료되지 않는 작업이 있을 경우, 다른 노드에서 동일한 작업을 시작하고, 먼저 완료되는 task를 성공처리하고, 이외의 다른 시도는 종료한다.

클러스터 내 노드 간 성능 차이가 있는 경우 speculative execution 기능을 활성화하는 것이 좋다.

speculative execution

 

주요 구성 옵션

• spark.speculation (default=false)
  true 설정 시 speculative execution을 실행한다.
  도입시기: 0.6.0
• spark.speculation.interval (default=100ms)
  spark가 speculate 할 작업을 확인하는 빈도
  도입시기: 0.6.0
• spark.speculation.multiplier (default = 1.5)
  task 실행시간의 중앙값보다 얼마나 느릴 때 speculative execution이 고려되어야 하는지에 대한 값
  도입시기: 0.6.0
• spark.speculation.quantile (default = 0.75)
  speculation이 활성화되기 전에 완료되어야 하는 task의 비율
  도입시기: 0.6.0

 

spark.speculation이 활성화 된 후 각 주요 구성 옵션의 기본값을 해석해보면:

• 매 100ms 마다 (spark.speculation.interval) spark는 느리게 실행되는 작업을 확인한다.
• 완료된 작업의 실행시간의 중앙값 * 1.5 < target task 실행시간 일 때 느리게 실행되는 작업으로 간주한다. (spark.speculation.multiplier)
• 스파크는 전체 예약된 작업의 75%가 완료되기 전까지는 speculation을 실행하지 않는다. (spark.speculation.quantile)

https://kb.databricks.com/scala/understanding-speculative-execution

speculative execution이 활성화되고 느린 작업으로 간주된 task의 경우 상태가 success인 작업과 taskKilled인 작업 두 개가 표시된다.

 

speculative execution을 활성화하면 좋은 경우

• <몇 가지 작업때문에 전체 실행시간이 너무 길어지는 경우>
  일부 작업이 너무 오랫동안 실행되고 원인이 확인되지 않은 경우에는 speculative execution을 사용할 수 있다.
  그러나 근본적인 원인이 확인되면 해결하고 비활성화 해야한다.
• <잘못된 VM 인스턴스가 있는 경우>
  speculative exectuion은 원래 작업과 동일한 노드에서 예약되지 않으므로 특정 vm이 비정상적이거나 task를 실행하기에 적합하지 않은 노드가 스파크 클러스터의 일부라면 해당 옵션을 활성화하여 완화할 수 있다.

 

speculative exection을 활성화하면 안좋은 경우

• 프로덕션 작업일 때
  프로덕션 작업에서 speculation을 장기간 사용하면 작업의 실패 확률이 높아진다.
• 작업이 멱등성이 아닌 경우
  추측실행을 활성화 하면 retry 후 result가 달라질 수 있다.
• skewed data로 인해 오래 걸리는 작업의 경우
  skewed data의 경우 speculative execution task가 원래 작업만큼 오래 걸릴 수도 있다.
  speculative exectuion은 원래 작업보다 먼저 완료된다는 것을 보장하지 않으므로 활성화해봤자 원래 작업이 성공하고 speculative execution이 종료될 가능성이 있다.
• 성능 영향
  활성화하면 성능에 영향을 미칠 수 있기 때문에 작업 속도 저하 시 단기적인 문제 해결용으로만 사용해야 한다.

 

추가 구성 옵션

timethreshold = (spark.speculation.multiplier * successfulTaskDurations.median) or (spark.speculation.minTaskRunTime) 로 정의

• spark.speculation.minTaskRuntime (defaul = 100ms)
  speculation을 고려하기 전에 작업이 실행되는 최소 시간.
  매우 짧은 task의 speculative execution을 방지하기 위해 사용된다.
  도입시기: 3.2.0
• spark.speculation.efficiency.processRateMultiplier (default = 0.75)
  비효율적인 작업을 평가할 때 사용되는 multiplier.
  값이 높을 수록 더 많은 task가 비효율적인 작업으로 간주될 수 있다.
  `calculator.getRunningTasksProcessRate(tid) <
                calculator.getAvgTaskProcessRate() * efficientTaskProcessMultiplier`
  도입시기: 3.4.0
• spark.speculation.efficiency.longRunTaskFactor (default = 2)
  task 내의 process rate 양호 여부와 상관없이 길게 실행되는 Task 누락을방지하기 위해 설정하는 값
  spark.speculation.efficiency.longRunFactor * time threshold을 초과하는 작업은 speculative 작업이 실행된다.
  도입시기: 3.4.0
• spark.speculation.efficiency.enabled (default = true)
  true 설정 시
  1) processing rate가 성공한 모든 task의 평균 * multiplier 보다 작거나
  2) duration이 spark.speculation.efficiency.longRunTaskFactor * time thishold을 초과한 경우
  비효율 적인 task로 판단 후 speculate한다.

 

주의

speculative execution은 execution time을 개선하기 위해 리소스가 더 많이 투입되는 실행 방식이며,

만약 Speculative Task 자체가 비효율적인 Task(Slow Task)가 된다면 리소스도 쓰고 시간적 이점도 없는 상황이 초래될 수 있따.

 

참고

https://spark.apache.org/docs/latest/configuration.html#scheduling

https://kb.databricks.com/scala/understanding-speculative-execution

https://github.com/apache/spark/blob/9b1b2b30d5919c4de358e82786dc631a3ab914ea/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L1208