我试图用 TotalOrderPartitioner 做一个简单的排序示例。输入是一个序列文件，以 IntWritable 为键，以 NullWritable 为值。我想根据键排序。的输出是一个以 IntWritable 为键，以 NullWritable 为值的序列文件。我在集群环境中运行这项工作。这是我的驱动程序类:

public class SortDriver extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = this.getConf();

        Job job = Job.getInstance(conf);
        job.setJobName("SORT-WITH-TOTAL-ORDER-PARTITIONER");
        job.setJarByClass(SortDriver.class);
        job.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.setInputPaths(job, new Path("/user/client/seq-input"));
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);
        job.setPartitionerClass(TotalOrderPartitioner.class);
        TotalOrderPartitioner.setPartitionFile(conf, new Path("/user/client/partition.lst"));
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setCompressOutput(job, true);
        SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);
        SequenceFileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
        SequenceFileOutputFormat.setOutputPath(job, new Path("/user/client/sorted-output"));
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(NullWritable.class);
        job.setNumReduceTasks(3);

        InputSampler.Sampler<IntWritable, NullWritable> sampler = new InputSampler.RandomSampler<>(0.1, 200);
        InputSampler.writePartitionFile(job, sampler);

        boolean res = job.waitForCompletion(true);

        return res ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new SortDriver(), args));
    }
}

映射器类:

public class SortMapper extends Mapper<IntWritable, NullWritable, IntWritable, NullWritable>{

    @Override
    protected void map(IntWritable key, NullWritable value, Context context) throws IOException, InterruptedException {
        context.write(key, value);
    }
}

Reducer 类:

public class SortReducer extends Reducer<IntWritable, NullWritable, IntWritable, NullWritable> {

    @Override
    protected void reduce(IntWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        context.write(key, NullWritable.get());
    }
}

当我运行这个作业时，我得到:

Error: java.lang.IllegalArgumentException: Can't read partitions file
    at org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner.setConf(TotalOrderPartitioner.java:116)
    at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:73)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:133)
    at org.apache.hadoop.mapred.MapTask$NewOutputCollector.<init>(MapTask.java:678)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:747)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:167)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1557)
    at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:162)
Caused by: java.io.FileNotFoundException: File file:/grid/hadoop/yarn/local/usercache/client/appcache/application_1406784047304_0002/container_1406784047304_0002_01_000003/_partition.lst does not exist
    at org.apache.hadoop.fs.RawLocalFileSystem.deprecatedGetFileStatus(RawLocalFileSystem.java:511)
    at org.apache.hadoop.fs.RawLocalFileSystem.getFileLinkStatusInternal(RawLocalFileSystem.java:724)
    at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:501)
    at org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:397)
    at org.apache.hadoop.io.SequenceFile$Reader.<init>(SequenceFile.java:1749)
    at org.apache.hadoop.io.SequenceFile$Reader.<init>(SequenceFile.java:1773)
    at org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner.readPartitions(TotalOrderPartitioner.java:301)
    at org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner.setConf(TotalOrderPartitioner.java:88)
    ... 10 more

我在主目录 (/user/client) 中找到名为 _partition.lst 的分区文件。分区文件名与代码不匹配:TotalOrderPartitioner.setPartitionFile(conf, new Path("/user/client/partition.lst"));。谁能帮我解决这个问题？我在 HDP 2.1 发行版中使用 hadoop 2.4。

最佳答案

我认为问题在于:

TotalOrderPartitioner.setPartitionFile(conf, new Path("/user/client/partition.lst"));

您必须将其替换为:

TotalOrderPartitioner.setPartitionFile(job.getConfiguration(), new Path("/user/client/partition.lst"));

因为你正在使用

InputSampler.writePartitionFile(job, sampler);

否则，请尝试仅将最后一行替换为:

InputSampler.writePartitionFile(conf, sampler);

但我不确定它在新 API 中是否会像那样工作。

希望对您有所帮助!祝你好运!

关于java - TotalOrderPartitioner 忽略分区文件位置，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/25051671/