org.apache.giraph.master

Class BspServiceMaster<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.master.BspServiceMaster<I,V,E>

Type Parameters:

I - Vertex id

V - Vertex data

E - Edge data

All Implemented Interfaces:

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

public class BspServiceMaster<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 CentralizedServiceMaster<I,V,E>, ResetSuperstepMetricsObserver

ZooKeeper-based implementation of CentralizedServiceMaster.

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 int	DEFAULT_INPUT_SPLIT_THREAD_COUNT Default number of threads to use when writing input splits to zookeeper
static int	MAX_PRINTABLE_REMAINING_WORKERS Print worker names only if there are 10 workers left
static String	NUM_MASTER_ZK_INPUT_SPLIT_THREADS How many threads to use when writing input splits to zookeeper

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
BspServiceMaster(org.apache.hadoop.mapreduce.Mapper.Context context, GraphTaskManager<I,V,E> graphTaskManager) Constructor for setting up the master.

Method Summary

Modifier and Type	Method and Description
void	addGiraphTimersAndSendCounters(long superstep) We add the Giraph Timers separately, because we need to include the time required for shutdown and cleanup This will fetch the final Giraph Timers, and send all the counters to the job client
boolean	becomeMaster() Become the master.
List<WorkerInfo>	checkWorkers() Check all the WorkerInfo objects to ensure that a minimum number of good workers exists out of the total that have reported.
void	cleanup(SuperstepState superstepState) Clean up the service (no calls may be issued after this)
SuperstepState	coordinateSuperstep() Master coordinates the superstep
int	createEdgeInputSplits() Create the BspInputSplit objects from the index range based on the user-defined EdgeInputFormat.
int	createMappingInputSplits() Create the BspInputSplit objects from the index range based on the user-defined MappingInputFormat.
int	createVertexInputSplits() Create the BspInputSplit objects from the index range based on the user-defined VertexInputFormat.
void	failureCleanup(Exception e) Called when the job fails in order to let the Master do any cleanup.
AggregatorToGlobalCommTranslation	getAggregatorTranslationHandler() Handler for aggregators to reduce/broadcast translation
MasterGlobalCommHandler	getGlobalCommHandler() Get handler for global communication
long	getLastGoodCheckpoint() Get the last known good checkpoint
MasterCompute	getMasterCompute() Get MasterCompute object
MasterInfo	getMasterInfo() Get master info
BspEvent	getSuperstepStateChangedEvent() Event that the master watches that denotes if a worker has done something that changes the state of a superstep (either a worker completed or died)
List<WorkerInfo>	getWorkerInfoList() Get list of workers
BspEvent	getWorkerWroteCheckpointEvent() Event that the master watches that denotes when a worker wrote checkpoint
void	newSuperstep(SuperstepMetricsRegistry superstepMetrics) Starting a new superstep.
void	postApplication() Application has finished.
void	postSuperstep() Superstep has finished.
boolean	processEvent(org.apache.zookeeper.WatchedEvent event) Derived classes that want additional ZooKeeper events to take action should override this.
void	restartFromCheckpoint(long checkpoint) Master can decide to restart from the last good checkpoint if a worker fails during a superstep.
void	setJobState(ApplicationState state, long applicationAttempt, long desiredSuperstep) If the master decides that this job doesn't have the resources to continue, it can fail the job.
void	setup() Setup (must be called prior to any other function)

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.CentralizedService

getConfiguration, getJobProgressTracker, getRestartedSuperstep, getSuperstep

Field Detail

MAX_PRINTABLE_REMAINING_WORKERS

public static final int MAX_PRINTABLE_REMAINING_WORKERS

Print worker names only if there are 10 workers left

See Also:

Constant Field Values

NUM_MASTER_ZK_INPUT_SPLIT_THREADS

public static final String NUM_MASTER_ZK_INPUT_SPLIT_THREADS

How many threads to use when writing input splits to zookeeper

See Also:

Constant Field Values

DEFAULT_INPUT_SPLIT_THREAD_COUNT

public static final int DEFAULT_INPUT_SPLIT_THREAD_COUNT

Default number of threads to use when writing input splits to zookeeper

See Also:

Constant Field Values

Constructor Detail

BspServiceMaster

public BspServiceMaster(org.apache.hadoop.mapreduce.Mapper.Context context,
                        GraphTaskManager<I,V,E> graphTaskManager)

Constructor for setting up the master.

Parameters:

context - Mapper context

graphTaskManager - GraphTaskManager for this compute node

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.

setJobState

public void setJobState(ApplicationState state,
                        long applicationAttempt,
                        long desiredSuperstep)

Description copied from interface: CentralizedServiceMaster

If the master decides that this job doesn't have the resources to continue, it can fail the job. It can also designate what to do next. Typically this is mainly informative.

Specified by:

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

Parameters:

state - State of the application.

applicationAttempt - Attempt to start on

desiredSuperstep - Superstep to restart from (if applicable)

checkWorkers

public List<WorkerInfo> checkWorkers()

Description copied from interface: CentralizedServiceMaster

Check all the WorkerInfo objects to ensure that a minimum number of good workers exists out of the total that have reported.

Specified by:

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

Returns:

List of of healthy workers such that the minimum has been met, otherwise null

createMappingInputSplits

public int createMappingInputSplits()

Description copied from interface: CentralizedServiceMaster

Create the BspInputSplit objects from the index range based on the user-defined MappingInputFormat. The BspInputSplit objects will processed by the workers later on during the INPUT_SUPERSTEP.

Specified by:

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

Returns:

Number of splits. Returns -1 on failure to create valid input splits.

createVertexInputSplits

public int createVertexInputSplits()

Description copied from interface: CentralizedServiceMaster

Create the BspInputSplit objects from the index range based on the user-defined VertexInputFormat. The BspInputSplit objects will processed by the workers later on during the INPUT_SUPERSTEP.

Specified by:

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

Returns:

Number of splits. Returns -1 on failure to create valid input splits.

createEdgeInputSplits

public int createEdgeInputSplits()

Description copied from interface: CentralizedServiceMaster

Create the BspInputSplit objects from the index range based on the user-defined EdgeInputFormat. The BspInputSplit objects will processed by the workers later on during the INPUT_SUPERSTEP.

Specified by:

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

Returns:

Number of splits. Returns -1 on failure to create valid input splits.

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

getGlobalCommHandler

public MasterGlobalCommHandler getGlobalCommHandler()

Description copied from interface: CentralizedServiceMaster

Get handler for global communication

Specified by:

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

Returns:

Global communication handler

getAggregatorTranslationHandler

public AggregatorToGlobalCommTranslation getAggregatorTranslationHandler()

Description copied from interface: CentralizedServiceMaster

Handler for aggregators to reduce/broadcast translation

Specified by:

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

Returns:

aggregator translation handler

getMasterCompute

public MasterCompute getMasterCompute()

Description copied from interface: CentralizedServiceMaster

Get MasterCompute object

Specified by:

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

Returns:

MasterCompute object

setup

public void setup()

Description copied from interface: CentralizedServiceMaster

Setup (must be called prior to any other function)

Specified by:

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

becomeMaster

public boolean becomeMaster()

Description copied from interface: CentralizedServiceMaster

Become the master.

Specified by:

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

Returns:

true if became the master, false if the application is done.

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

restartFromCheckpoint

public void restartFromCheckpoint(long checkpoint)

Description copied from interface: CentralizedServiceMaster

Master can decide to restart from the last good checkpoint if a worker fails during a superstep.

Specified by:

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

Parameters:

checkpoint - Checkpoint to restart from

getLastGoodCheckpoint

public long getLastGoodCheckpoint()
                           throws IOException

Description copied from interface: CentralizedServiceMaster

Get the last known good checkpoint

Specified by:

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

Returns:

Last good superstep number

Throws:

IOException

coordinateSuperstep

public SuperstepState coordinateSuperstep()
                                   throws org.apache.zookeeper.KeeperException,
                                          InterruptedException

Description copied from interface: CentralizedServiceMaster

Master coordinates the superstep

Specified by:

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

Returns:

State of the application as a result of this superstep

Throws:

org.apache.zookeeper.KeeperException

InterruptedException

addGiraphTimersAndSendCounters

public void addGiraphTimersAndSendCounters(long superstep)

We add the Giraph Timers separately, because we need to include the time required for shutdown and cleanup This will fetch the final Giraph Timers, and send all the counters to the job client

Specified by:

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

Parameters:

superstep - superstep for which the GiraphTimer will be sent

postApplication

public void postApplication()

Description copied from interface: CentralizedServiceMaster

Application has finished.

Specified by:

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

postSuperstep

public void postSuperstep()

Description copied from interface: CentralizedServiceMaster

Superstep has finished.

Specified by:

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

failureCleanup

public void failureCleanup(Exception e)

Description copied from interface: CentralizedServiceMaster

Called when the job fails in order to let the Master do any cleanup.

Specified by:

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

Parameters:

e - Exception job failed from. May be null.

cleanup

public void cleanup(SuperstepState superstepState)
             throws IOException

Description copied from interface: CentralizedServiceMaster

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

Specified by:

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

Parameters:

superstepState - what was the state of the last complete superstep?

Throws:

IOException

getWorkerWroteCheckpointEvent

public final BspEvent getWorkerWroteCheckpointEvent()

Event that the master watches that denotes when a worker wrote checkpoint

Returns:

Event that denotes when a worker wrote checkpoint

getSuperstepStateChangedEvent

public final BspEvent getSuperstepStateChangedEvent()

Event that the master watches that denotes if a worker has done something that changes the state of a superstep (either a worker completed or died)

Returns:

Event that denotes a superstep state change

processEvent

public 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