EpochCoordinator-spark技术分享

EpochCoordinator

EpochCoordinator is a ThreadSafeRpcEndpoint that receive messages from…FIXME

EpochCoordinator is created (using create factory method) when ContinuousExecution is requested to run a streaming query in continuous mode.

Table 1. EpochCoordinator RPC Endpoint’s Messages
Message	Description
`CommitPartitionEpoch`
`GetCurrentEpoch`
`IncrementAndGetEpoch`
`ReportPartitionOffset`
`SetReaderPartitions`
`SetWriterPartitions`
`StopContinuousExecutionWrites`

Table 2. EpochCoordinator’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`queryWritesStopped`	Flag that…FIXME Used when…FIXME

Tip

Enable INFO logging level for org.apache.spark.sql.execution.streaming.continuous.EpochCoordinatorRef* logger to see what happens inside.

Add the following line to conf/log4j.properties:



log4j.logger.org.apache.spark.sql.execution.streaming.continuous.EpochCoordinatorRef*=INFO

log4j.logger.org.apache.spark.sql.execution.streaming.continuous.EpochCoordinatorRef*=INFO

Refer to Logging.

Receiving Messages — `receive` Method



receive: PartialFunction[Any, Unit]

receive: PartialFunction[Any, Unit]

Note	`receive` is part of the `RpcEndpoint` Contract in Apache Spark to receive messages.

receive…FIXME

`resolveCommitsAtEpoch` Internal Method



resolveCommitsAtEpoch(epoch: Long): Unit

resolveCommitsAtEpoch(epoch: Long): Unit

resolveCommitsAtEpoch…FIXME

Note	`resolveCommitsAtEpoch` is used exclusively when `EpochCoordinator` is requested to receive `CommitPartitionEpoch` and `ReportPartitionOffset` messages.

`commitEpoch` Internal Method



commitEpoch(
  epoch: Long,
  messages: Iterable[WriterCommitMessage]): Unit

commitEpoch(

epoch: Long,

messages: Iterable[WriterCommitMessage]): Unit

commitEpoch…FIXME

Note	`commitEpoch` is used exclusively when `EpochCoordinator` is requested to resolveCommitsAtEpoch.

Creating EpochCoordinator Instance

EpochCoordinator takes the following to be created:

StreamWriter
ContinuousReader
ContinuousExecution
Start epoch
SparkSession
RpcEnv

EpochCoordinator initializes the internal registries and counters.

Registering EpochCoordinator RPC Endpoint — `create` Factory Method



create(
  writer: StreamWriter,
  reader: ContinuousReader,
  query: ContinuousExecution,
  epochCoordinatorId: String,
  startEpoch: Long,
  session: SparkSession,
  env: SparkEnv): RpcEndpointRef

create(

writer: StreamWriter,

reader: ContinuousReader,

query: ContinuousExecution,

epochCoordinatorId: String,

startEpoch: Long,

session: SparkSession,

env: SparkEnv): RpcEndpointRef

create simply creates a new EpochCoordinator and requests the RpcEnv to register a RPC endpoint as EpochCoordinator-[id] (where id is the given epochCoordinatorId).

create prints out the following INFO message to the logs:



Registered EpochCoordinator endpoint

Registered EpochCoordinator endpoint

Note	`create` is used exclusively when `ContinuousExecution` is requested to run a streaming query in continuous mode.

EpochCoordinator