in src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java [2365:2614]
private Pair<MapReduceOper,Integer> getSamplingJob(POSort sort, MapReduceOper prevJob, List<PhysicalPlan> transformPlans,
FileSpec lFile, FileSpec sampleFile, int rp, List<PhysicalPlan> sortKeyPlans,
String udfClassName, String[] udfArgs, String sampleLdrClassName ) throws PlanException, VisitorException {
String[] rslargs = new String[2];
// SampleLoader expects string version of FuncSpec
// as its first constructor argument.
rslargs[0] = (new FuncSpec(Utils.getTmpFileCompressorName(pigContext))).toString();
// This value is only used by order by. For skewed join, it's calculated
// based on the file size.
rslargs[1] = pigContext.getProperties().getProperty(PigConfiguration.PIG_RANDOM_SAMPLER_SAMPLE_SIZE, "100");
FileSpec quantLdFilName = new FileSpec(lFile.getFileName(),
new FuncSpec(sampleLdrClassName, rslargs));
MapReduceOper mro = startNew(quantLdFilName, prevJob);
if(sort.isUDFComparatorUsed) {
mro.UDFs.add(sort.getMSortFunc().getFuncSpec().toString());
curMROp.isUDFComparatorUsed = true;
}
List<Boolean> flat1 = new ArrayList<Boolean>();
List<PhysicalPlan> eps1 = new ArrayList<PhysicalPlan>();
// if transform plans are not specified, project the columns of sorting keys
if (transformPlans == null) {
Pair<POProject, Byte>[] sortProjs = null;
try{
sortProjs = getSortCols(sort.getSortPlans());
}catch(Exception e) {
throw new RuntimeException(e);
}
// Set up the projections of the key columns
if (sortProjs == null) {
PhysicalPlan ep = new PhysicalPlan();
POProject prj = new POProject(new OperatorKey(scope,
nig.getNextNodeId(scope)));
prj.setStar(true);
prj.setOverloaded(false);
prj.setResultType(DataType.TUPLE);
ep.add(prj);
eps1.add(ep);
flat1.add(false);
} else {
for (Pair<POProject, Byte> sortProj : sortProjs) {
// Check for proj being null, null is used by getSortCols for a non POProject
// operator. Since Order by does not allow expression operators,
//it should never be set to null
if(sortProj == null){
int errCode = 2174;
String msg = "Internal exception. Could not create a sampler job";
throw new MRCompilerException(msg, errCode, PigException.BUG);
}
PhysicalPlan ep = new PhysicalPlan();
POProject prj;
try {
prj = sortProj.first.clone();
} catch (CloneNotSupportedException e) {
//should not get here
throw new AssertionError(
"Error cloning project caught exception" + e
);
}
ep.add(prj);
eps1.add(ep);
flat1.add(false);
}
}
}else{
for(int i=0; i<transformPlans.size(); i++) {
eps1.add(transformPlans.get(i));
flat1.add(i == transformPlans.size() - 1 ? true : false);
}
}
// This foreach will pick the sort key columns from the RandomSampleLoader output
POForEach nfe1 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)),-1,eps1,flat1);
mro.mapPlan.addAsLeaf(nfe1);
// Now set up a POLocalRearrange which has "all" as the key and the output of the
// foreach will be the "value" out of POLocalRearrange
PhysicalPlan ep1 = new PhysicalPlan();
ConstantExpression ce = new ConstantExpression(new OperatorKey(scope,nig.getNextNodeId(scope)));
ce.setValue("all");
ce.setResultType(DataType.CHARARRAY);
ep1.add(ce);
List<PhysicalPlan> eps = new ArrayList<PhysicalPlan>();
eps.add(ep1);
POLocalRearrange lr = new POLocalRearrange(new OperatorKey(scope,nig.getNextNodeId(scope)));
try {
lr.setIndex(0);
} catch (ExecException e) {
int errCode = 2058;
String msg = "Unable to set index on newly created POLocalRearrange.";
throw new PlanException(msg, errCode, PigException.BUG, e);
}
lr.setKeyType(DataType.CHARARRAY);
lr.setPlans(eps);
lr.setResultType(DataType.TUPLE);
lr.addOriginalLocation(sort.getAlias(), sort.getOriginalLocations());
mro.mapPlan.add(lr);
mro.mapPlan.connect(nfe1, lr);
mro.setMapDone(true);
POPackage pkg = new POPackage(new OperatorKey(scope,nig.getNextNodeId(scope)));
Packager pkgr = new Packager();
pkg.setPkgr(pkgr);
pkgr.setKeyType(DataType.CHARARRAY);
pkg.setNumInps(1);
boolean[] inner = {false};
pkgr.setInner(inner);
mro.reducePlan.add(pkg);
// Lets start building the plan which will have the sort
// for the foreach
PhysicalPlan fe2Plan = new PhysicalPlan();
// Top level project which just projects the tuple which is coming
// from the foreach after the package
POProject topPrj = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
topPrj.setColumn(1);
topPrj.setResultType(DataType.BAG);
topPrj.setOverloaded(true);
fe2Plan.add(topPrj);
// the projections which will form sort plans
List<PhysicalPlan> nesSortPlanLst = new ArrayList<PhysicalPlan>();
if (sortKeyPlans != null) {
for(int i=0; i<sortKeyPlans.size(); i++) {
nesSortPlanLst.add(sortKeyPlans.get(i));
}
}else{
Pair<POProject, Byte>[] sortProjs = null;
try{
sortProjs = getSortCols(sort.getSortPlans());
}catch(Exception e) {
throw new RuntimeException(e);
}
// Set up the projections of the key columns
if (sortProjs == null) {
PhysicalPlan ep = new PhysicalPlan();
POProject prj = new POProject(new OperatorKey(scope,
nig.getNextNodeId(scope)));
prj.setStar(true);
prj.setOverloaded(false);
prj.setResultType(DataType.TUPLE);
ep.add(prj);
nesSortPlanLst.add(ep);
} else {
for (int i=0; i<sortProjs.length; i++) {
POProject prj =
new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
prj.setResultType(sortProjs[i].second);
if(sortProjs[i].first != null && sortProjs[i].first.isProjectToEnd()){
if(i != sortProjs.length -1){
//project to end has to be the last sort column
throw new AssertionError("Project-range to end (x..)" +
" is supported in order-by only as last sort column");
}
prj.setProjectToEnd(i);
break;
}
else{
prj.setColumn(i);
}
prj.setOverloaded(false);
PhysicalPlan ep = new PhysicalPlan();
ep.add(prj);
nesSortPlanLst.add(ep);
}
}
}
sort.setSortPlans(nesSortPlanLst);
sort.setResultType(DataType.BAG);
fe2Plan.add(sort);
fe2Plan.connect(topPrj, sort);
// The plan which will have a constant representing the
// degree of parallelism for the final order by map-reduce job
// this will either come from a "order by parallel x" in the script
// or will be the default number of reducers for the cluster if
// "parallel x" is not used in the script
PhysicalPlan rpep = new PhysicalPlan();
ConstantExpression rpce = new ConstantExpression(new OperatorKey(scope,nig.getNextNodeId(scope)));
rpce.setRequestedParallelism(rp);
// We temporarily set it to rp and will adjust it at runtime, because the final degree of parallelism
// is unknown until we are ready to submit it. See PIG-2779.
rpce.setValue(rp);
rpce.setResultType(DataType.INTEGER);
rpep.add(rpce);
List<PhysicalPlan> genEps = new ArrayList<PhysicalPlan>();
genEps.add(rpep);
genEps.add(fe2Plan);
List<Boolean> flattened2 = new ArrayList<Boolean>();
flattened2.add(false);
flattened2.add(false);
POForEach nfe2 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)),-1, genEps, flattened2);
mro.reducePlan.add(nfe2);
mro.reducePlan.connect(pkg, nfe2);
// Let's connect the output from the foreach containing
// number of quantiles and the sorted bag of samples to
// another foreach with the FindQuantiles udf. The input
// to the FindQuantiles udf is a project(*) which takes the
// foreach input and gives it to the udf
PhysicalPlan ep4 = new PhysicalPlan();
POProject prjStar4 = new POProject(new OperatorKey(scope,nig.getNextNodeId(scope)));
prjStar4.setResultType(DataType.TUPLE);
prjStar4.setStar(true);
ep4.add(prjStar4);
List<PhysicalOperator> ufInps = new ArrayList<PhysicalOperator>();
ufInps.add(prjStar4);
POUserFunc uf = new POUserFunc(new OperatorKey(scope,nig.getNextNodeId(scope)), -1, ufInps,
new FuncSpec(udfClassName, udfArgs));
ep4.add(uf);
ep4.connect(prjStar4, uf);
List<PhysicalPlan> ep4s = new ArrayList<PhysicalPlan>();
ep4s.add(ep4);
List<Boolean> flattened3 = new ArrayList<Boolean>();
flattened3.add(false);
POForEach nfe3 = new POForEach(new OperatorKey(scope,nig.getNextNodeId(scope)), -1, ep4s, flattened3);
mro.reducePlan.add(nfe3);
mro.reducePlan.connect(nfe2, nfe3);
POStore str = getStore();
str.setSFile(sampleFile);
mro.reducePlan.add(str);
mro.reducePlan.connect(nfe3, str);
mro.setReduceDone(true);
mro.requestedParallelism = 1;
mro.markSampler();
return new Pair<MapReduceOper, Integer>(mro, rp);
}