BarrierTaskContext — TaskContext for Barrier Tasks¶

BarrierTaskContext is a concrete TaskContext of the tasks in a Barrier Stage in Barrier Execution Mode.

Creating Instance¶

BarrierTaskContext takes the following to be created:

• TaskContext

BarrierTaskContext is created when:

• Task is requested to run (with isBarrier flag enabled)

Barrier Coordinator RPC Endpoint¶

barrierCoordinator: RpcEndpointRef

BarrierTaskContext creates a RpcEndpointRef to Barrier Coordinator RPC Endpoint when created.

barrierCoordinator is used to handle barrier and allGather operators (through runBarrier).

allGather¶

allGather(
  message: String): Array[String]

allGather runBarrier with the given message and ALL_GATHER request method.

barrier¶

barrier(): Unit

barrier runBarrier with no message and BARRIER request method.

Global Sync¶

runBarrier(
  message: String,
  requestMethod: RequestMethod.Value): Array[String]

runBarrier prints out the following INFO message to the logs:

Task [taskAttemptId] from Stage [stageId](Attempt [stageAttemptNumber]) has entered the global sync, current barrier epoch is [barrierEpoch].

runBarrier prints out the following TRACE message to the logs:

Current callSite: [callSite]

runBarrier schedules a TimerTask (Java) to print out the following INFO message to the logs every minute:

Task [taskAttemptId] from Stage [stageId](Attempt [stageAttemptNumber]) waiting under the global sync since [startTime],
has been waiting for [duration] seconds,
current barrier epoch is [barrierEpoch].

runBarrier requests the Barrier Coordinator RPC Endpoint to send a RequestToSync one-off message and waits 365 days (!) for a response (a collection of responses from all the barrier tasks).

runBarrier checks every second if the response "bundle" arrived.

runBarrier increments the barrierEpoch.

runBarrier prints out the following INFO message to the logs:

Task [taskAttemptId] from Stage [stageId](Attempt [stageAttemptNumber]) finished global sync successfully,
waited for [duration] seconds,
current barrier epoch is [barrierEpoch].

In the end, runBarrier returns the response "bundle" (a collection of responses from all the barrier tasks).

In case of a SparkException, runBarrier prints out the following INFO message to the logs and reports (re-throws) the exception up (the call chain):

Task [taskAttemptId] from Stage [stageId](Attempt [stageAttemptNumber]) failed to perform global sync,
waited for [duration] seconds,
current barrier epoch is [barrierEpoch].

runBarrier is used when:

• BarrierTaskContext is requested to barrier, allGather

Logging¶

Enable ALL logging level for org.apache.spark.BarrierTaskContext logger to see what happens inside.

Add the following line to conf/log4j2.properties:

logger.BarrierTaskContext.name = org.apache.spark.BarrierTaskContext
logger.BarrierTaskContext.level = all

Refer to Logging.