org.apache.giraph.block_app.migration

Class MigrationPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,MPrev extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable>

java.lang.Object

org.apache.giraph.block_app.framework.piece.AbstractPiece<I,V,E,M,WV,WM,S>

org.apache.giraph.block_app.framework.piece.DefaultParentPiece<I,V,E,M,WV,WM,S>

org.apache.giraph.block_app.framework.piece.PieceWithWorkerContext<I,V,E,M,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

org.apache.giraph.block_app.migration.MigrationPiece<I,V,E,MPrev,M>

Type Parameters:

I - Vertex id type

V - Vertex value type

E - Edge value type

MPrev - Previous piece message type

M - Message type

All Implemented Interfaces:

Iterable<AbstractPiece>, Block

public final class MigrationPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,MPrev extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable>
extends PieceWithWorkerContext<I,V,E,M,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

Piece used when migrating applications to Blocks Framework. There are two migration levels:

• drop-in replacement migration is completely compatible with previous code. Only necessary thing is to change parent classes from (AbstractComputation, MasterCompute, WorkerContext) to (MigrationFullAbstractComputation, MigrationFullMasterCompute and MigrationFullWorkerContext). After that, all you need to do is extend MigrationBlockFactory, and pass appropriate types and call createMigrationAppBlock with initial computations.
  You can now combine multiple applications, or use any library written in the framework, but your application is left as one whole indivisible block.
• Piece-wise migration - which gives a set of independent pieces, which can then be combined with appropriate ordering logic within a BlockFactory. You need to modify parent classes in your code to (MigrationAbstractComputation, MigrationMasterCompute and MigrationWorkerContext), which don't have any methods that affect computation ordering - and so calling those methods should be moved to logic within BlockFactory. Calling MigrationPiece.createMigrationPiece and passing appropriate computations, gives you an independent piece, that you can then use in the same way as before, but also combine it in any other way with other pieces you have or are written within a library.

Generally, migration path can be to first move to drop-in replacement without any effort, and then see which parts need to be modified to be able to use piece-wise migration. At the end, it should be trivial to move from piece-wise migration to directly using pieces, by just moving code around, if you want to.

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.giraph.block_app.framework.piece.AbstractPiece

AbstractPiece.InnerVertexReceiver, AbstractPiece.InnerVertexSender

Field Summary

Fields inherited from class org.apache.giraph.block_app.framework.piece.DefaultParentPiece

MESSAGE_ENCODE_AND_STORE_TYPE_MIN_FORCE

Method Summary

Modifier and Type	Method and Description
static <I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,MR extends org.apache.hadoop.io.Writable,MS extends org.apache.hadoop.io.Writable> MigrationPiece<I,V,E,MR,MS>	createMigrationPiece(Class<? extends MigrationAbstractComputation<I,V,E,MR,MS>> computationClass, MigrationMasterCompute masterCompute, Supplier<Iterable<MR>> previousMessagesSupplier, Consumer<Iterable<MS>> currentMessagesConsumer, Class<MS> messageClass, Class<? extends MessageCombiner<? super I,MS>> messageCombinerClass)
MessageClasses<I,M>	getMessageClasses(ImmutableClassesGiraphConfiguration conf) Returns MessageClasses definition for messages being sent by this Piece.
MigrationPiece	getNextPiece()
VertexReceiver<I,V,E,M>	getVertexReceiver(BlockWorkerReceiveApi<I> workerApi, MigrationSuperstepStage executionStage) Override to do vertex receive processing.
VertexSender<I,V,E>	getVertexSender(BlockWorkerSendApi<I,V,E,M> workerApi, MigrationSuperstepStage executionStage) Override to do vertex send processing.
void	masterCompute(BlockMasterApi masterApi, MigrationSuperstepStage executionStage) Function that is called on master, after send phase, before receive phase.
MigrationSuperstepStage	nextExecutionStage(MigrationSuperstepStage executionStage) Override to provide different next execution stage for Pieces that come after it.
void	registerAggregators(BlockMasterApi masterApi) Override to register any potential aggregators used by this piece.
void	workerContextReceive(BlockWorkerContextReceiveApi workerContextApi, MigrationSuperstepStage executionStage, MigrationWorkerContext workerValue, List<org.apache.hadoop.io.Writable> workerMessages) Override to have worker context receive computation.
void	workerContextSend(BlockWorkerContextSendApi<I,org.apache.hadoop.io.Writable> workerContextApi, MigrationSuperstepStage executionStage, MigrationWorkerContext workerValue) Override to have worker context send computation.

Methods inherited from class org.apache.giraph.block_app.framework.piece.PieceWithWorkerContext

getWorkerValue

Methods inherited from class org.apache.giraph.block_app.framework.piece.DefaultParentPiece

allowOneMessageToManyIdsEncoding, getMessageClass, getMessageCombiner, getMessageFactory, getWrappedVertexSender, receiveIgnoreExistingVertices, reduceDouble, reduceFloat, reduceInt, reduceLong, registerReducers, wrappedRegisterReducers

Methods inherited from class org.apache.giraph.block_app.framework.piece.AbstractPiece

equals, forAllPossiblePieces, getPieceCount, hashCode, iterator, toString

Methods inherited from class java.lang.Object

clone, finalize, getClass, notify, notifyAll, wait, wait, wait

Methods inherited from interface java.lang.Iterable

forEach, spliterator

Method Detail

createMigrationPiece

public static <I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,MR extends org.apache.hadoop.io.Writable,MS extends org.apache.hadoop.io.Writable> MigrationPiece<I,V,E,MR,MS> createMigrationPiece(Class<? extends MigrationAbstractComputation<I,V,E,MR,MS>> computationClass,
                                                                                                                                                                                                                                                                                     MigrationMasterCompute masterCompute,
                                                                                                                                                                                                                                                                                     Supplier<Iterable<MR>> previousMessagesSupplier,
                                                                                                                                                                                                                                                                                     Consumer<Iterable<MS>> currentMessagesConsumer,
                                                                                                                                                                                                                                                                                     Class<MS> messageClass,
                                                                                                                                                                                                                                                                                     Class<? extends MessageCombiner<? super I,MS>> messageCombinerClass)

registerAggregators

public void registerAggregators(BlockMasterApi masterApi)
                         throws InstantiationException,
                                IllegalAccessException

Description copied from class: AbstractPiece

Override to register any potential aggregators used by this piece.

Overrides:

registerAggregators in class AbstractPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

Throws:

InstantiationException

IllegalAccessException

getVertexSender

public VertexSender<I,V,E> getVertexSender(BlockWorkerSendApi<I,V,E,M> workerApi,
                                           MigrationSuperstepStage executionStage)

Description copied from class: DefaultParentPiece

Override to do vertex send processing. Creates handler that defines what should be executed on worker during send phase. This logic gets executed first. This function is called once on each worker on each thread, in parallel, on their copy of Piece object to create functions handler. If returned object implements Postprocessor interface, then corresponding postprocess() function is going to be called once, after all vertices corresponding thread needed to process are done.

Overrides:

getVertexSender in class DefaultParentPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

workerContextSend

public void workerContextSend(BlockWorkerContextSendApi<I,org.apache.hadoop.io.Writable> workerContextApi,
                              MigrationSuperstepStage executionStage,
                              MigrationWorkerContext workerValue)

Description copied from class: AbstractPiece

Override to have worker context send computation. Called once per worker, after all vertices have been processed with getVertexSender.

Overrides:

workerContextSend in class AbstractPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

masterCompute

public void masterCompute(BlockMasterApi masterApi,
                          MigrationSuperstepStage executionStage)

Description copied from class: AbstractPiece

Function that is called on master, after send phase, before receive phase. It can: - read aggregators sent from worker - do global processing - send data to workers through aggregators

Overrides:

masterCompute in class AbstractPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

workerContextReceive

public void workerContextReceive(BlockWorkerContextReceiveApi workerContextApi,
                                 MigrationSuperstepStage executionStage,
                                 MigrationWorkerContext workerValue,
                                 List<org.apache.hadoop.io.Writable> workerMessages)

Description copied from class: AbstractPiece

Override to have worker context receive computation. Called once per worker, before all vertices are going to be processed with getVertexReceiver.

Overrides:

workerContextReceive in class AbstractPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

getVertexReceiver

public VertexReceiver<I,V,E,M> getVertexReceiver(BlockWorkerReceiveApi<I> workerApi,
                                                 MigrationSuperstepStage executionStage)

Description copied from class: AbstractPiece

Override to do vertex receive processing. Creates handler that defines what should be executed on worker for each vertex during receive phase. This logic executed last. This function is called once on each worker on each thread, in parallel, on their copy of Piece object to create functions handler. If returned object implements Postprocessor interface, then corresponding postprocess() function is going to be called once, after all vertices corresponding thread needed to process are done.

Overrides:

getVertexReceiver in class AbstractPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

getMessageClasses

public MessageClasses<I,M> getMessageClasses(ImmutableClassesGiraphConfiguration conf)

Description copied from class: AbstractPiece

Returns MessageClasses definition for messages being sent by this Piece.

Overrides:

getMessageClasses in class DefaultParentPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

nextExecutionStage

public MigrationSuperstepStage nextExecutionStage(MigrationSuperstepStage executionStage)

Description copied from class: AbstractPiece

Override to provide different next execution stage for Pieces that come after it. Execution stage should be immutable, and this function should be returning a new object, if it needs to return different value. It affects pieces that come after this piece, and isn't applied to execution stage this piece sees.

Overrides:

nextExecutionStage in class AbstractPiece<I extends org.apache.hadoop.io.WritableComparable,V extends org.apache.hadoop.io.Writable,E extends org.apache.hadoop.io.Writable,M extends org.apache.hadoop.io.Writable,MigrationWorkerContext,org.apache.hadoop.io.Writable,MigrationSuperstepStage>

getNextPiece

public MigrationPiece getNextPiece()