org.apache.pig.backend.hadoop.executionengine.mapReduceLayer

Class MRCompiler

java.lang.Object

org.apache.pig.impl.plan.PlanVisitor<PhysicalOperator,PhysicalPlan>

org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor

org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRCompiler

public class MRCompiler
extends PhyPlanVisitor

The compiler that compiles a given physical plan into a DAG of MapReduce operators which can then be converted into the JobControl structure. Is implemented as a visitor of the PhysicalPlan it is compiling. Currently supports all operators except the MR Sort operator Uses a predecessor based depth first traversal. To compile an operator, first compiles the predecessors into MapReduce Operators and tries to merge the current operator into one of them. The goal being to keep the number of MROpers to a minimum. It also merges multiple Map jobs, created by compiling the inputs individually, into a single job. Here a new map job is created and then the contents of the previous map plans are added. However, any other state that was in the previous map plans, should be manually moved over. So, if you are adding something new take care about this. Ex of this is in requestedParallelism Only in case of blocking operators and splits, a new MapReduce operator is started using a store-load combination to connect the two operators. Whenever this happens care is taken to add the MROper into the MRPlan and connect it appropriately.

Field Summary

Fields

Modifier and Type	Field and Description
static String	FILE_CONCATENATION_THRESHOLD
static String	OPTIMISTIC_FILE_CONCATENATION
static String	USER_COMPARATOR_MARKER

Fields inherited from class org.apache.pig.impl.plan.PlanVisitor

mCurrentWalker, mPlan

Constructor Summary

Constructors

Constructor and Description
MRCompiler(PhysicalPlan plan)
MRCompiler(PhysicalPlan plan, PigContext pigContext)

Method Summary

Methods

Modifier and Type	Method and Description
void	aggregateScalarsFiles()
MROperPlan	compile() The front-end method that the user calls to compile the plan.
void	connectSoftLink()
CompilationMessageCollector	getMessageCollector()
MROperPlan	getMRPlan() Used to get the compiled plan
PhysicalPlan	getPlan() Used to get the plan that was compiled
void	visitCollectedGroup(POCollectedGroup op)
void	visitCounter(POCounter op) For the counter job, it depends if it is row number or not.
void	visitCross(POCross op)
void	visitDistinct(PODistinct op)
void	visitFilter(POFilter op)
void	visitFRJoin(POFRJoin op) This is an operator which will have multiple inputs(= to number of join inputs) But it prunes off all inputs but the fragment input and creates separate MR jobs for each of the replicated inputs and uses these as the replicated files that are configured in the POFRJoin operator.
void	visitGlobalRearrange(POGlobalRearrange op)
void	visitLimit(POLimit op)
void	visitLoad(POLoad op)
void	visitLocalRearrange(POLocalRearrange op)
void	visitMergeCoGroup(POMergeCogroup poCoGrp) Leftmost relation is referred as base relation (this is the one fed into mappers.) First, close all MROpers except for first one (referred as baseMROPer) Then, create a MROper which will do indexing job (idxMROper) Connect idxMROper before the mappedMROper in the MRPlan.
void	visitMergeJoin(POMergeJoin joinOp) Since merge-join works on two inputs there are exactly two MROper predecessors identified as left and right.
void	visitNative(PONative op)
void	visitPackage(POPackage op)
void	visitPOForEach(POForEach op)
void	visitRank(PORank op) In case of PORank, it is closed any other previous job (containing POCounter as a leaf) and PORank is added on map phase.
void	visitSkewedJoin(POSkewedJoin op)
void	visitSort(POSort op)
void	visitSplit(POSplit op) Compiles a split operator.
void	visitStore(POStore op)
void	visitStream(POStream op)
void	visitUnion(POUnion op)

Methods inherited from class org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor

visit, visitAdd, visitAnd, visitBinCond, visitCast, visitComparisonFunc, visitConstant, visitDemux, visitDivide, visitEqualTo, visitGreaterThan, visitGTOrEqual, visitIsNull, visitLessThan, visitLTOrEqual, visitMapLookUp, visitMod, visitMultiply, visitNegative, visitNot, visitNotEqualTo, visitOr, visitPartialAgg, visitPartitionRearrange, visitPoissonSample, visitPOOptimizedForEach, visitPreCombinerLocalRearrange, visitProject, visitRegexp, visitReservoirSample, visitSubtract, visitUserFunc

Methods inherited from class org.apache.pig.impl.plan.PlanVisitor

popWalker, pushWalker, visit

Methods inherited from class java.lang.Object

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

Field Detail

USER_COMPARATOR_MARKER

public static final String USER_COMPARATOR_MARKER

See Also:

Constant Field Values

FILE_CONCATENATION_THRESHOLD

public static final String FILE_CONCATENATION_THRESHOLD

See Also:

Constant Field Values

OPTIMISTIC_FILE_CONCATENATION

public static final String OPTIMISTIC_FILE_CONCATENATION

See Also:

Constant Field Values

Constructor Detail

MRCompiler

public MRCompiler(PhysicalPlan plan)
           throws MRCompilerException

Throws:

MRCompilerException

MRCompiler

public MRCompiler(PhysicalPlan plan,
          PigContext pigContext)
           throws MRCompilerException

Throws:

MRCompilerException

Method Detail

aggregateScalarsFiles

public void aggregateScalarsFiles()
                           throws PlanException,
                                  IOException

Throws:

PlanException

IOException

getMRPlan

public MROperPlan getMRPlan()

Used to get the compiled plan

Returns:

map reduce plan built by the compiler

getPlan

public PhysicalPlan getPlan()

Used to get the plan that was compiled

Overrides:

getPlan in class PlanVisitor<PhysicalOperator,PhysicalPlan>

Returns:

physical plan

getMessageCollector

public CompilationMessageCollector getMessageCollector()

compile

public MROperPlan compile()
                   throws IOException,
                          PlanException,
                          VisitorException

The front-end method that the user calls to compile the plan. Assumes that all submitted plans have a Store operators as the leaf.

Returns:

A map reduce plan

Throws:

IOException

PlanException

VisitorException

connectSoftLink

public void connectSoftLink()
                     throws PlanException,
                            IOException

Throws:

PlanException

IOException

visitSplit

public void visitSplit(POSplit op)
                throws VisitorException

Compiles a split operator. The logic is to close the split job by replacing the split oper by a store and creating a new Map MRoper and return that as the current MROper to which other operators would be compiled into. The new MROper would be connected to the split job by load-store. Also add the split oper to the splitsSeen map.

Overrides:

visitSplit in class PhyPlanVisitor

Parameters:

op - - The split operator

Throws:

VisitorException

visitLoad

public void visitLoad(POLoad op)
               throws VisitorException

Overrides:

visitLoad in class PhyPlanVisitor

Throws:

VisitorException

visitNative

public void visitNative(PONative op)
                 throws VisitorException

Overrides:

visitNative in class PhyPlanVisitor

Throws:

VisitorException

visitStore

public void visitStore(POStore op)
                throws VisitorException

Overrides:

visitStore in class PhyPlanVisitor

Throws:

VisitorException

visitFilter

public void visitFilter(POFilter op)
                 throws VisitorException

Overrides:

visitFilter in class PhyPlanVisitor

Throws:

VisitorException

visitCross

public void visitCross(POCross op)
                throws VisitorException

Overrides:

visitCross in class PhyPlanVisitor

Throws:

VisitorException

visitStream

public void visitStream(POStream op)
                 throws VisitorException

Overrides:

visitStream in class PhyPlanVisitor

Throws:

VisitorException

visitLimit

public void visitLimit(POLimit op)
                throws VisitorException

Overrides:

visitLimit in class PhyPlanVisitor

Throws:

VisitorException

visitLocalRearrange

public void visitLocalRearrange(POLocalRearrange op)
                         throws VisitorException

Overrides:

visitLocalRearrange in class PhyPlanVisitor

Throws:

VisitorException

visitCollectedGroup

public void visitCollectedGroup(POCollectedGroup op)
                         throws VisitorException

Overrides:

visitCollectedGroup in class PhyPlanVisitor

Throws:

VisitorException

visitPOForEach

public void visitPOForEach(POForEach op)
                    throws VisitorException

Overrides:

visitPOForEach in class PhyPlanVisitor

Throws:

VisitorException

visitGlobalRearrange

public void visitGlobalRearrange(POGlobalRearrange op)
                          throws VisitorException

Overrides:

visitGlobalRearrange in class PhyPlanVisitor

Throws:

VisitorException

visitPackage

public void visitPackage(POPackage op)
                  throws VisitorException

Overrides:

visitPackage in class PhyPlanVisitor

Throws:

VisitorException

visitUnion

public void visitUnion(POUnion op)
                throws VisitorException

Overrides:

visitUnion in class PhyPlanVisitor

Throws:

VisitorException

visitFRJoin

public void visitFRJoin(POFRJoin op)
                 throws VisitorException

This is an operator which will have multiple inputs(= to number of join inputs) But it prunes off all inputs but the fragment input and creates separate MR jobs for each of the replicated inputs and uses these as the replicated files that are configured in the POFRJoin operator. It also sets that this is FRJoin job and some parametes associated with it.

Overrides:

visitFRJoin in class PhyPlanVisitor

Throws:

VisitorException

visitMergeCoGroup

public void visitMergeCoGroup(POMergeCogroup poCoGrp)
                       throws VisitorException

Leftmost relation is referred as base relation (this is the one fed into mappers.) First, close all MROpers except for first one (referred as baseMROPer) Then, create a MROper which will do indexing job (idxMROper) Connect idxMROper before the mappedMROper in the MRPlan.

Overrides:

visitMergeCoGroup in class PhyPlanVisitor

Throws:

VisitorException

visitMergeJoin

public void visitMergeJoin(POMergeJoin joinOp)
                    throws VisitorException

Since merge-join works on two inputs there are exactly two MROper predecessors identified as left and right. Instead of merging two operators, both are used to generate a MR job each. First MR oper is run to generate on-the-fly index on right side. Second is used to actually do the join. First MR oper is identified as rightMROper and second as curMROper. 1) RightMROper: If it is in map phase. It can be preceded only by POLoad. If there is anything else in physical plan, that is yanked and set as inner plans of joinOp. If it is reduce phase. Close this operator and start new MROper. 2) LeftMROper: If it is in map phase, add the Join operator in it. If it is in reduce phase. Close it and start new MROper.

Overrides:

visitMergeJoin in class PhyPlanVisitor

Throws:

VisitorException

visitDistinct

public void visitDistinct(PODistinct op)
                   throws VisitorException

Overrides:

visitDistinct in class PhyPlanVisitor

Throws:

VisitorException

visitSkewedJoin

public void visitSkewedJoin(POSkewedJoin op)
                     throws VisitorException

Overrides:

visitSkewedJoin in class PhyPlanVisitor

Throws:

VisitorException

visitSort

public void visitSort(POSort op)
               throws VisitorException

Overrides:

visitSort in class PhyPlanVisitor

Throws:

VisitorException

visitCounter

public void visitCounter(POCounter op)
                  throws VisitorException

For the counter job, it depends if it is row number or not. In case of being a row number, any previous jobs are saved and POCounter is added as a leaf on a map task. If it is not, then POCounter is added as a leaf on a reduce task (last sorting phase).

Overrides:

visitCounter in class PhyPlanVisitor

Throws:

VisitorException

visitRank

public void visitRank(PORank op)
               throws VisitorException

In case of PORank, it is closed any other previous job (containing POCounter as a leaf) and PORank is added on map phase.

Overrides:

visitRank in class PhyPlanVisitor

Throws:

VisitorException