java - 无法在 Map Side Join 中使用 Composite textinputFormat

Question

我正在尝试使用 CompositeTextInoutFormat 实现 Map-side join。但是，我在 Map reduce 作业中遇到以下错误，我无法解决。1. 在下面的代码中，我在使用 Compose 方法时出错，在设置 inputformat 类时也出错。错误如下所示。

The method compose(String, Class, Path...) in the type CompositeInputFormat is not applicable for the arguments (String, Class, Path[])

谁能帮忙

package Hadoop.MR.Practice;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.join.CompositeInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
//import org.apache.hadoop.mapred.join.CompositeInputFormat;


public class MapJoinJob implements Tool{

private Configuration conf;     

public Configuration getConf() {
    return conf;
}
public void setConf(Configuration conf) {
    this.conf = conf;
}
@Override
public int run(String[] args) throws Exception {
    Job job = Job.getInstance(getConf(), "MapSideJoinJob");
    job.setJarByClass(this.getClass());

    Path[] inputs = new Path[] { new Path(args[0]), new Path(args[1])};
    String join = CompositeInputFormat.compose("inner", KeyValueTextInputFormat.class, inputs);
    job.getConfiguration().set("mapreduce.join.expr", join);

    job.setInputFormatClass(CompositeInputFormat.class);

    job.setMapperClass(MapJoinMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(LongWritable.class);

    //Configuring reducer
    job.setReducerClass(WCReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);
    job.setNumReduceTasks(0);

    FileOutputFormat.setOutputPath(job, new Path(args[2]));

    job.waitForCompletion(true);
    return 0;
}

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    MapJoinJob mjJob = new MapJoinJob();
    ToolRunner.run(conf, mjJob, args);
}

Answer 1

我会说您的问题可能与混合 hadoop API 有关。您可以看到您的导入混合了 mapred 和 mapreduce。

例如，您尝试将 org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat 与 org.apache.hadoop.mapred.join.CompositeInputFormat 一起使用> 这不太可能奏效。

您应该选择一个(我会说可能是 mapreduce)并确保一切都使用相同的 API。