org.apache.giraph.worker

Class BspServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

java.lang.Object

org.apache.giraph.bsp.BspService<I,V,E>

org.apache.giraph.worker.BspServiceWorker<I,V,E>

Type Parameters:

I - Vertex id

V - Vertex data

E - Edge data

All Implemented Interfaces:

CentralizedService<I,V,E>, CentralizedServiceWorker<I,V,E>, PartitionSplitInfo<I>, ResetSuperstepMetricsObserver, org.apache.zookeeper.Watcher

public class BspServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>
extends BspService<I,V,E>
implements CentralizedServiceWorker<I,V,E>, ResetSuperstepMetricsObserver

ZooKeeper-based implementation of CentralizedServiceWorker.

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.zookeeper.Watcher

org.apache.zookeeper.Watcher.Event

Field Summary

Fields

Modifier and Type	Field and Description
static String	TIMER_WAIT_REQUESTS Name of gauge for time spent waiting on other workers

Fields inherited from class org.apache.giraph.bsp.BspService

APPLICATION_ATTEMPTS_DIR, applicationAttemptsPath, BASE_DIR, basePath, checkpointBasePath, CLEANED_UP_DIR, cleanedUpPath, COUNTERS_DIR, FORCE_CHECKPOINT_USER_FLAG, HALT_COMPUTATION_NODE, haltComputationPath, INPUT_SPLITS_ALL_DONE_NODE, INPUT_SPLITS_WORKER_DONE_DIR, INPUT_SUPERSTEP, inputSplitsAllDonePath, inputSplitsWorkerDonePath, JSONOBJ_APPLICATION_ATTEMPT_KEY, JSONOBJ_METRICS_KEY, JSONOBJ_NUM_MESSAGE_BYTES_KEY, JSONOBJ_NUM_MESSAGES_KEY, JSONOBJ_STATE_KEY, JSONOBJ_SUPERSTEP_KEY, KRYO_REGISTERED_CLASS_DIR, kryoRegisteredClassPath, MASTER_ELECTION_DIR, MASTER_JOB_STATE_NODE, MASTER_SUFFIX, masterElectionPath, masterJobStatePath, MEMORY_OBSERVER_DIR, memoryObserverPath, METRICS_DIR, PARTITION_EXCHANGE_DIR, savedCheckpointBasePath, SUPERSTEP_DIR, SUPERSTEP_FINISHED_NODE, UNSET_APPLICATION_ATTEMPT, UNSET_SUPERSTEP, WORKER_FINISHED_DIR, WORKER_HEALTHY_DIR, WORKER_SUFFIX, WORKER_UNHEALTHY_DIR, WORKER_WROTE_CHECKPOINT_DIR

Constructor Summary

Constructors

Constructor and Description
BspServiceWorker(org.apache.hadoop.mapreduce.Mapper.Context context, GraphTaskManager<I,V,E> graphTaskManager) Constructor for setting up the worker.

Method Summary

Modifier and Type	Method and Description
void	addressesAndPartitionsReceived(AddressesAndPartitionsWritable addressesAndPartitions) Received addresses and partitions assignments from master.
void	cleanup(FinishedSuperstepStats finishedSuperstepStats) Clean up the service (no calls may be issued after this)
void	closeZooKeeper() Method to close the zookeeper connection, after the worker has sent the counters to the master
void	exchangeVertexPartitions(Collection<? extends PartitionOwner> masterSetPartitionOwners) If desired by the user, vertex partitions are redistributed among workers according to the chosen WorkerGraphPartitioner.
void	failureCleanup() Operations that will be called if there is a failure by a worker.
FinishedSuperstepStats	finishSuperstep(List<PartitionStats> partitionStatsList, GiraphTimerContext superstepTimerContext) Worker is done with its portion of the superstep.
WorkerAggregatorHandler	getAggregatorHandler() Get worker aggregator handler
GlobalStats	getGlobalStats() Loads Global stats from zookeeper.
WorkerInputSplitsHandler	getInputSplitsHandler() Get input splits handler used during input
LocalData<I,V,E,? extends org.apache.hadoop.io.Writable>	getLocalData()
MasterInfo	getMasterInfo() Get master info
Partition	getNextPartition() PartitionStore.getNextPartition()
BspEvent	getPartitionExchangeChildrenChangedEvent() Get event when the state of a partition exchange has changed.
int	getPartitionId(I vertexId) Get the partition id that a vertex id would belong to.
Iterable<Integer>	getPartitionIds() Get the ids of all the stored partitions (on current worker) as Iterable
Iterable<? extends PartitionOwner>	getPartitionOwners() Get all partition owners.
PartitionStore<I,V,E>	getPartitionStore() Get the partition store for this worker.
long	getPartitionVertexCount(Integer partitionId) Return the number of vertices in a partition.
ServerData<I,V,E>	getServerData() Get server data
SuperstepOutput<I,V,E>	getSuperstepOutput() Get the superstep output class
TranslateEdge<I,E>	getTranslateEdge()
PartitionOwner	getVertexPartitionOwner(I vertexId) Every client will need to get a partition owner from a vertex id so that they know which worker to sent the request to.
WorkerClient<I,V,E>	getWorkerClient() Get the worker client (for instantiating WorkerClientRequestProcessor instances.
WorkerContext	getWorkerContext() Get the worker context.
WorkerInfo	getWorkerInfo() Get the worker information
List<WorkerInfo>	getWorkerInfoList() Get list of workers
WorkerObserver[]	getWorkerObservers() Get the observers for this Worker.
boolean	hasPartition(Integer partitionId) Whether a partition with given id exists on this worker.
boolean	isHealthy() Intended to check the health of the node.
VertexEdgeCount	loadCheckpoint(long superstep) Load the vertices, edges, messages from the beginning of a superstep.
void	newSuperstep(SuperstepMetricsRegistry superstepMetrics) Starting a new superstep.
void	prepareSuperstep() Final preparation for superstep, called after startSuperstep and potential loading from checkpoint, right before the computation started TODO how to avoid this additional function
protected boolean	processEvent(org.apache.zookeeper.WatchedEvent event) Derived classes that want additional ZooKeeper events to take action should override this.
void	putPartition(Partition partition) PartitionStore.putPartition(Partition)
FinishedSuperstepStats	setup() Setup (must be called prior to any other function)
void	startIteration() PartitionStore.startIteration()
Collection<? extends PartitionOwner>	startSuperstep() Take all steps prior to actually beginning the computation of a superstep.
void	storeCheckpoint() Both the vertices and the messages need to be checkpointed in order for them to be used.
void	storeCountersInZooKeeper(boolean allSuperstepsDone) Method to store the counter values in the zookeeper This is called at the end of each superstep and after finishing all the supersteps

Methods inherited from class org.apache.giraph.bsp.BspService

getApplicationAttempt, getApplicationAttemptChangedEvent, getCheckpointBasePath, getCleanedUpChildrenChangedEvent, getConfiguration, getContext, getFs, getGraphPartitionerFactory, getGraphTaskManager, getHealthyHostnameIdFromPath, getHostname, getHostnameTaskId, getInputSplitsAllDoneEvent, getInputSplitsWorkerDoneEvent, getJobId, getJobProgressTracker, getJobState, getLastCheckpointedSuperstep, getMasterElectionChildrenChangedEvent, getPartitionExchangePath, getPartitionExchangeWorkerPath, getRestartedSuperstep, getSavedCheckpointBasePath, getSuperstep, getSuperstepFinishedEvent, getSuperstepFinishedPath, getSuperstepFromPath, getSuperstepPath, getTaskId, getWorkerCountersFinishedPath, getWorkerHealthRegistrationChangedEvent, getWorkerId, getWorkerInfoById, getWorkerInfoHealthyPath, getWorkerInfoUnhealthyPath, getWorkerMetricsFinishedPath, getWorkerWroteCheckpointPath, getWrittenCountersToZKEvent, getZkExt, incrCachedSuperstep, process, registerBspEvent, setApplicationAttempt, setCachedSuperstep, setRestartedSuperstep

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.giraph.bsp.CentralizedServiceWorker

getGraphTaskManager

Methods inherited from interface org.apache.giraph.bsp.CentralizedService

getConfiguration, getJobProgressTracker, getRestartedSuperstep, getSuperstep

Field Detail

TIMER_WAIT_REQUESTS

public static final String TIMER_WAIT_REQUESTS

Name of gauge for time spent waiting on other workers

See Also:

Constant Field Values

Constructor Detail

BspServiceWorker

public BspServiceWorker(org.apache.hadoop.mapreduce.Mapper.Context context,
                        GraphTaskManager<I,V,E> graphTaskManager)
                 throws IOException,
                        InterruptedException

Constructor for setting up the worker.

Parameters:

context - Mapper context

graphTaskManager - GraphTaskManager for this compute node

Throws:

IOException

InterruptedException

Method Detail

newSuperstep

public void newSuperstep(SuperstepMetricsRegistry superstepMetrics)

Description copied from interface: ResetSuperstepMetricsObserver

Starting a new superstep. Re-initialize your metrics.

Specified by:

newSuperstep in interface ResetSuperstepMetricsObserver

Parameters:

superstepMetrics - SuperstepMetricsRegistry being used.

getWorkerContext

public WorkerContext getWorkerContext()

Description copied from interface: CentralizedServiceWorker

Get the worker context.

Specified by:

getWorkerContext in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

worker's WorkerContext

getWorkerObservers

public WorkerObserver[] getWorkerObservers()

Description copied from interface: CentralizedServiceWorker

Get the observers for this Worker.

Specified by:

getWorkerObservers in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

array of WorkerObservers.

getWorkerClient

public WorkerClient<I,V,E> getWorkerClient()

Description copied from interface: CentralizedServiceWorker

Get the worker client (for instantiating WorkerClientRequestProcessor instances.

Specified by:

getWorkerClient in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Worker client

getLocalData

public LocalData<I,V,E,? extends org.apache.hadoop.io.Writable> getLocalData()

getTranslateEdge

public TranslateEdge<I,E> getTranslateEdge()

isHealthy

public boolean isHealthy()

Intended to check the health of the node. For instance, can it ssh, dmesg, etc. For now, does nothing. TODO: Make this check configurable by the user (i.e. search dmesg for problems).

Returns:

True if healthy (always in this case).

getMasterInfo

public MasterInfo getMasterInfo()

Description copied from interface: CentralizedService

Get master info

Specified by:

getMasterInfo in interface CentralizedService<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Master info

getWorkerInfoList

public List<WorkerInfo> getWorkerInfoList()

Description copied from interface: CentralizedService

Get list of workers

Specified by:

getWorkerInfoList in interface CentralizedService<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

List of workers

setup

public FinishedSuperstepStats setup()

Description copied from interface: CentralizedServiceWorker

Setup (must be called prior to any other function)

Specified by:

setup in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Finished superstep stats for the input superstep

failureCleanup

public void failureCleanup()

Description copied from interface: CentralizedServiceWorker

Operations that will be called if there is a failure by a worker.

Specified by:

failureCleanup in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

startSuperstep

public Collection<? extends PartitionOwner> startSuperstep()

Description copied from interface: CentralizedServiceWorker

Take all steps prior to actually beginning the computation of a superstep.

Specified by:

startSuperstep in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Collection of all the partition owners from the master for this superstep.

finishSuperstep

public FinishedSuperstepStats finishSuperstep(List<PartitionStats> partitionStatsList,
                                              GiraphTimerContext superstepTimerContext)

Description copied from interface: CentralizedServiceWorker

Worker is done with its portion of the superstep. Report the worker level statistics after the computation.

Specified by:

finishSuperstep in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

partitionStatsList - All the partition stats for this worker

superstepTimerContext - superstep timer context only given when the function needs to stop the timer, otherwise null.

Returns:

Stats of the superstep completion

cleanup

public void cleanup(FinishedSuperstepStats finishedSuperstepStats)
             throws IOException,
                    InterruptedException

Description copied from interface: CentralizedServiceWorker

Clean up the service (no calls may be issued after this)

Specified by:

cleanup in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

finishedSuperstepStats - Finished supestep stats

Throws:

IOException

InterruptedException

storeCountersInZooKeeper

public void storeCountersInZooKeeper(boolean allSuperstepsDone)

Method to store the counter values in the zookeeper This is called at the end of each superstep and after finishing all the supersteps

Specified by:

storeCountersInZooKeeper in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

allSuperstepsDone - Whether the job has finished all supersteps This is needed to ensure the superstep number is the same for master and worker, when all supersteps are finished

closeZooKeeper

public void closeZooKeeper()

Method to close the zookeeper connection, after the worker has sent the counters to the master

Specified by:

closeZooKeeper in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

storeCheckpoint

public void storeCheckpoint()
                     throws IOException

Description copied from interface: CentralizedServiceWorker

Both the vertices and the messages need to be checkpointed in order for them to be used. This is done after all messages have been delivered, but prior to a superstep starting.

Specified by:

storeCheckpoint in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Throws:

IOException

loadCheckpoint

public VertexEdgeCount loadCheckpoint(long superstep)

Description copied from interface: CentralizedServiceWorker

Load the vertices, edges, messages from the beginning of a superstep. Will load the vertex partitions as designated by the master and set the appropriate superstep.

Specified by:

loadCheckpoint in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

superstep - which checkpoint to use

Returns:

Graph-wide vertex and edge counts

exchangeVertexPartitions

public final void exchangeVertexPartitions(Collection<? extends PartitionOwner> masterSetPartitionOwners)

Description copied from interface: CentralizedServiceWorker

If desired by the user, vertex partitions are redistributed among workers according to the chosen WorkerGraphPartitioner.

Specified by:

exchangeVertexPartitions in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

masterSetPartitionOwners - Partition owner info passed from the master.

getPartitionExchangeChildrenChangedEvent

public final BspEvent getPartitionExchangeChildrenChangedEvent()

Get event when the state of a partition exchange has changed.

Returns:

Event to check.

processEvent

protected boolean processEvent(org.apache.zookeeper.WatchedEvent event)

Description copied from class: BspService

Derived classes that want additional ZooKeeper events to take action should override this.

Overrides:

processEvent in class BspService<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

event - Event that occurred

Returns:

true if the event was processed here, false otherwise

getWorkerInfo

public WorkerInfo getWorkerInfo()

Description copied from interface: CentralizedServiceWorker

Get the worker information

Specified by:

getWorkerInfo in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Worker information

getPartitionStore

public PartitionStore<I,V,E> getPartitionStore()

Description copied from interface: CentralizedServiceWorker

Get the partition store for this worker. The partitions contain the vertices for this worker and can be used to run compute() for the vertices or do checkpointing.

Specified by:

getPartitionStore in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

The partition store for this worker.

getVertexPartitionOwner

public PartitionOwner getVertexPartitionOwner(I vertexId)

Description copied from interface: CentralizedServiceWorker

Every client will need to get a partition owner from a vertex id so that they know which worker to sent the request to.

Specified by:

getVertexPartitionOwner in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

vertexId - Vertex index to look for

Returns:

PartitionOnwer that should contain this vertex if it exists

getPartitionOwners

public Iterable<? extends PartitionOwner> getPartitionOwners()

Description copied from interface: CentralizedServiceWorker

Get all partition owners.

Specified by:

getPartitionOwners in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Iterable through partition owners

getPartitionId

public int getPartitionId(I vertexId)

Description copied from interface: CentralizedServiceWorker

Get the partition id that a vertex id would belong to.

Specified by:

getPartitionId in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Specified by:

getPartitionId in interface PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable>

Parameters:

vertexId - Vertex id

Returns:

Partition id

hasPartition

public boolean hasPartition(Integer partitionId)

Description copied from interface: CentralizedServiceWorker

Whether a partition with given id exists on this worker.

Specified by:

hasPartition in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

partitionId - Partition id

Returns:

True iff this worker has the specified partition

getPartitionIds

public Iterable<Integer> getPartitionIds()

Description copied from interface: PartitionSplitInfo

Get the ids of all the stored partitions (on current worker) as Iterable

Specified by:

getPartitionIds in interface PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable>

Returns:

The partition ids

getPartitionVertexCount

public long getPartitionVertexCount(Integer partitionId)

Description copied from interface: PartitionSplitInfo

Return the number of vertices in a partition.

Specified by:

getPartitionVertexCount in interface PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable>

Parameters:

partitionId - Partition id

Returns:

The number of vertices in the specified partition

startIteration

public void startIteration()

Description copied from interface: PartitionSplitInfo

PartitionStore.startIteration()

Specified by:

startIteration in interface PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable>

getNextPartition

public Partition getNextPartition()

Description copied from interface: PartitionSplitInfo

PartitionStore.getNextPartition()

Specified by:

getNextPartition in interface PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable>

Returns:

The next partition to process

putPartition

public void putPartition(Partition partition)

Description copied from interface: PartitionSplitInfo

PartitionStore.putPartition(Partition)

Specified by:

putPartition in interface PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable>

Parameters:

partition - Partition

getServerData

public ServerData<I,V,E> getServerData()

Description copied from interface: CentralizedServiceWorker

Get server data

Specified by:

getServerData in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Server data

getAggregatorHandler

public WorkerAggregatorHandler getAggregatorHandler()

Description copied from interface: CentralizedServiceWorker

Get worker aggregator handler

Specified by:

getAggregatorHandler in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Worker aggregator handler

prepareSuperstep

public void prepareSuperstep()

Description copied from interface: CentralizedServiceWorker

Final preparation for superstep, called after startSuperstep and potential loading from checkpoint, right before the computation started TODO how to avoid this additional function

Specified by:

prepareSuperstep in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

getSuperstepOutput

public SuperstepOutput<I,V,E> getSuperstepOutput()

Description copied from interface: CentralizedServiceWorker

Get the superstep output class

Specified by:

getSuperstepOutput in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

SuperstepOutput

getGlobalStats

public GlobalStats getGlobalStats()

Description copied from interface: CentralizedServiceWorker

Loads Global stats from zookeeper.

Specified by:

getGlobalStats in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

global stats stored in zookeeper for previous superstep.

getInputSplitsHandler

public WorkerInputSplitsHandler getInputSplitsHandler()

Description copied from interface: CentralizedServiceWorker

Get input splits handler used during input

Specified by:

getInputSplitsHandler in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Returns:

Input splits handler

addressesAndPartitionsReceived

public void addressesAndPartitionsReceived(AddressesAndPartitionsWritable addressesAndPartitions)

Description copied from interface: CentralizedServiceWorker

Received addresses and partitions assignments from master.

Specified by:

addressesAndPartitionsReceived in interface CentralizedServiceWorker<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable>

Parameters:

addressesAndPartitions - Addresses and partitions assignment