in src/org/apache/pig/backend/hadoop/executionengine/tez/plan/optimizer/ParallelismSetter.java [72:200]
public void visitTezOp(TezOperator tezOp) throws VisitorException {
if (tezOp instanceof NativeTezOper) {
return;
}
try {
// Can only set parallelism here if the parallelism isn't derived from
// splits
int parallelism = -1;
if (tezOp.getLoaderInfo().getLoads() != null && tezOp.getLoaderInfo().getLoads().size() > 0) {
// requestedParallelism of Loader vertex is handled in LoaderProcessor
// propogate to vertexParallelism
tezOp.setVertexParallelism(tezOp.getRequestedParallelism());
incrementTotalParallelism(tezOp, tezOp.getRequestedParallelism());
return;
} else {
int prevParallelism = -1;
boolean isOneToOneParallelism = false;
for (Map.Entry<OperatorKey,TezEdgeDescriptor> entry : tezOp.inEdges.entrySet()) {
if (entry.getValue().dataMovementType == DataMovementType.ONE_TO_ONE) {
TezOperator pred = mPlan.getOperator(entry.getKey());
parallelism = pred.getEffectiveParallelism(pc.defaultParallel);
if (prevParallelism == -1) {
prevParallelism = parallelism;
} else if (prevParallelism != parallelism) {
throw new VisitorException("one to one sources parallelism for vertex "
+ tezOp.getOperatorKey().toString() + " are not equal");
}
tezOp.setRequestedParallelism(pred.getRequestedParallelism());
// If tezOp.estimatedParallelism already set, don't override
// The only case is in PigGraceShuffleVertexManager, which
// set the estimated parallelism according to the output data size of the node
if (tezOp.getEstimatedParallelism()==-1) {
tezOp.setEstimatedParallelism(pred.getEstimatedParallelism());
}
isOneToOneParallelism = true;
incrementTotalParallelism(tezOp, parallelism);
parallelism = -1;
}
}
if (!isOneToOneParallelism) {
if (tezOp.getRequestedParallelism() != -1) {
parallelism = tezOp.getRequestedParallelism();
} else if (pc.defaultParallel != -1) {
parallelism = pc.defaultParallel;
}
if (parallelism == 0) {
// We need to produce empty output file.
// Even if user set PARALLEL 0, mapreduce has 1 reducer
parallelism = 1;
}
boolean overrideRequestedParallelism = false;
if (parallelism != -1
&& autoParallelismEnabled
&& !tezOp.isDontEstimateParallelism()
&& tezOp.isIntermediateReducer()
&& tezOp.isOverrideIntermediateParallelism()) {
overrideRequestedParallelism = true;
}
if (parallelism == -1 || overrideRequestedParallelism) {
if (tezOp.getEstimatedParallelism() == -1) {
// Override user specified parallelism with the estimated value
// if it is intermediate reducer
parallelism = estimator.estimateParallelism(mPlan, tezOp, conf);
if (overrideRequestedParallelism) {
if (tezOp.getRequestedParallelism() != parallelism) {
LOG.info("Increased requested parallelism of " + tezOp.getOperatorKey() + " to " + parallelism);
}
tezOp.setRequestedParallelism(parallelism);
} else {
tezOp.setEstimatedParallelism(parallelism);
}
} else {
parallelism = tezOp.getEstimatedParallelism();
}
if (tezOp.isGlobalSort() || tezOp.isSkewedJoin()) {
boolean additionalEdge = false;
if (tezOp.isGlobalSort() && getPlan().getPredecessors(tezOp).size() != 1 ||
tezOp.isSkewedJoin() && getPlan().getPredecessors(tezOp).size() != 2) {
additionalEdge = true;
}
if (!overrideRequestedParallelism && !additionalEdge) {
incrementTotalParallelism(tezOp, parallelism);
// PartitionerDefinedVertexManager will determine parallelism.
// So call setVertexParallelism with -1
// setEstimatedParallelism still needs to have some positive value
// so that TezDAGBuilder sets the PartitionerDefinedVertexManager
parallelism = -1;
} else {
// We are overriding the parallelism. We need to update the
// Constant value in sampleAggregator to same parallelism
// Currently will happen when you have orderby or
// skewed join followed by group by with combiner
for (TezOperator pred : mPlan.getPredecessors(tezOp)) {
if (pred.isSampleBasedPartitioner()) {
for (TezOperator partitionerPred : mPlan.getPredecessors(pred)) {
if (partitionerPred.isSampleAggregation() && partitionerPred.plan!=null) {
LOG.debug("Updating parallelism constant value to " + parallelism + " in " + partitionerPred.plan);
ParallelConstantVisitor visitor =
new ParallelConstantVisitor(partitionerPred.plan, parallelism);
visitor.visit();
partitionerPred.setNeedEstimatedQuantile(false);
break;
}
}
break;
}
}
}
}
}
}
}
incrementTotalParallelism(tezOp, parallelism);
tezOp.setVertexParallelism(parallelism);
// TODO: Fix case where vertex parallelism is -1 for auto parallelism with PartitionerDefinedVertexManager.
// i.e order by or skewed join followed by cross
if (tezOp.getCrossKeys() != null) {
for (String key : tezOp.getCrossKeys()) {
pc.getProperties().put(PigImplConstants.PIG_CROSS_PARALLELISM + "." + key,
Integer.toString(tezOp.getVertexParallelism()));
}
}
} catch (Exception e) {
throw new VisitorException(e);
}
}