diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/PartitionKeySampler.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/PartitionKeySampler.java index 96f4530..dc1b601 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/PartitionKeySampler.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/PartitionKeySampler.java @@ -112,7 +112,7 @@ public void collect(HiveKey key, Object value) throws IOException { return partitionKeys; } - public void writePartitionKeys(Path path, HiveConf conf, JobConf job) throws IOException { + public void writePartitionKeys(Path path, JobConf job) throws IOException { byte[][] partitionKeys = getPartitionKeys(job.getNumReduceTasks()); int numPartition = partitionKeys.length + 1; if (numPartition != job.getNumReduceTasks()) { @@ -133,10 +133,11 @@ public void writePartitionKeys(Path path, HiveConf conf, JobConf job) throws IOE } // random sampling - public static FetchOperator createSampler(FetchWork work, HiveConf conf, JobConf job, + public static FetchOperator createSampler(FetchWork work, JobConf job, Operator operator) throws HiveException { - int sampleNum = conf.getIntVar(HiveConf.ConfVars.HIVESAMPLINGNUMBERFORORDERBY); - float samplePercent = conf.getFloatVar(HiveConf.ConfVars.HIVESAMPLINGPERCENTFORORDERBY); + int sampleNum = HiveConf.getIntVar(job, HiveConf.ConfVars.HIVESAMPLINGNUMBERFORORDERBY); + float samplePercent = + HiveConf.getFloatVar(job, HiveConf.ConfVars.HIVESAMPLINGPERCENTFORORDERBY); if (samplePercent < 0.0 || samplePercent > 1.0) { throw new IllegalArgumentException("Percentile value must be within the range of 0 to 1."); } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java index d7a08ec..41d0493 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java @@ -376,7 +376,7 @@ public int execute(DriverContext driverContext) { if (mWork.getSamplingType() > 0 && rWork != null && job.getNumReduceTasks() > 1) { try { - handleSampling(driverContext, mWork, job, conf); + handleSampling(ctx, mWork, job); job.setPartitionerClass(HiveTotalOrderPartitioner.class); } catch (IllegalStateException e) { console.printInfo("Not enough sampling data.. Rolling back to single reducer task"); @@ -494,7 +494,7 @@ public int execute(DriverContext driverContext) { return (returnVal); } - private void handleSampling(DriverContext context, MapWork mWork, JobConf job, HiveConf conf) + private void handleSampling(Context context, MapWork mWork, JobConf job) throws Exception { assert mWork.getAliasToWork().keySet().size() == 1; @@ -510,7 +510,7 @@ private void handleSampling(DriverContext context, MapWork mWork, JobConf job, H inputPaths.add(new Path(path)); } - Path tmpPath = context.getCtx().getExternalTmpPath(inputPaths.get(0)); + Path tmpPath = context.getExternalTmpPath(inputPaths.get(0)); Path partitionFile = new Path(tmpPath, ".partitions"); ShimLoader.getHadoopShims().setTotalOrderPartitionFile(job, partitionFile); PartitionKeySampler sampler = new PartitionKeySampler(); @@ -539,9 +539,9 @@ private void handleSampling(DriverContext context, MapWork mWork, JobConf job, H fetchWork.setSource(ts); // random sampling - FetchOperator fetcher = PartitionKeySampler.createSampler(fetchWork, conf, job, ts); + FetchOperator fetcher = PartitionKeySampler.createSampler(fetchWork, job, ts); try { - ts.initialize(conf, new ObjectInspector[]{fetcher.getOutputObjectInspector()}); + ts.initialize(job, new ObjectInspector[]{fetcher.getOutputObjectInspector()}); OperatorUtils.setChildrenCollector(ts.getChildOperators(), sampler); while (fetcher.pushRow()) { } } finally { @@ -550,7 +550,7 @@ private void handleSampling(DriverContext context, MapWork mWork, JobConf job, H } else { throw new IllegalArgumentException("Invalid sampling type " + mWork.getSamplingType()); } - sampler.writePartitionKeys(partitionFile, conf, job); + sampler.writePartitionKeys(partitionFile, job); } /**