Hadoop Mapper 中未调用设置方法

Question

我运行了一系列 Hadoop Mapper/Reducer 并获得了电影 ID 列表。我使用 MovieData 文件根据这些 ID 显示电影的名称。我正在使用如下的 Mapper 类。我看到 setUp 方法没有被调用，因为我没有看到 print 语句，并且当我尝试使用加载方法中加载的这个 HashMap 时，我得到一个 Null 异常。以下是代码。任何指针表示赞赏。

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Mapper.Context;

public class MovieNamesMapper extends MapReduceBase implements Mapper<Object, Text, Text, Text> {

    private static HashMap<String, String> movieNameHashMap = new HashMap<String, String>();
    private BufferedReader bufferedReader;
    private String movieId = "";

    protected void setup(Context context) throws IOException,
            InterruptedException {

        System.out.println("Setting up system..");

        Path[] cacheFilesLocal = DistributedCache.getLocalCacheFiles(context
                .getConfiguration());

        for (Path eachPath : cacheFilesLocal) {
            if (eachPath.getName().toString().trim().equals("u.item")) {
                loadMovieNamesHashMap(eachPath, context);
            }
        }

    }

    private void loadMovieNamesHashMap(Path filePath, Context context)
            throws IOException {

        System.out.println("Loading movie names..");

        String strLineRead = "";

        try {
            bufferedReader = new BufferedReader(new FileReader(
                    filePath.toString()));

            while ((strLineRead = bufferedReader.readLine()) != null) {
                String movieIdArray[] = strLineRead.toString().split("\t|::");
                movieNameHashMap.put(movieIdArray[0].trim(),
                        movieIdArray[1].trim());
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (bufferedReader != null) {
                bufferedReader.close();

            }

        }

    }

    public void map(Object key, Text value, OutputCollector<Text, Text> output,
            Reporter reporter) throws IOException {
        System.out.println(key.toString() + " - " + value.toString());
        if (value.toString().length() > 0) {
            String moviePairArray[] = value.toString().split(":");

            for (String moviePair : moviePairArray) {
                String movieArray[] = moviePair.split(",");
                output.collect(new Text(movieNameHashMap.get(movieArray[0])),
                        new Text(movieNameHashMap.get(movieArray[1])));
            }
        }

    }

    public String getMovieId() {
        return movieId;
    }

    public void setMovieId(String movieId) {
        this.movieId = movieId;
    }

}

以下是我的运行方法。

public int run(String[] args) throws Exception {

    // For finding user and his rated movie list.
    JobConf conf1 = new JobConf(MovieTopDriver.class);
    conf1.setMapperClass(MoviePairsMapper.class);
    conf1.setReducerClass(MoviePairsReducer.class);

    conf1.setJarByClass(MovieTopDriver.class);

    FileInputFormat.addInputPath(conf1, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf1, new Path("temp"));

    conf1.setMapOutputKeyClass(Text.class);
    conf1.setMapOutputValueClass(Text.class);

    conf1.setOutputKeyClass(Text.class);
    conf1.setOutputValueClass(IntWritable.class);

    // For finding movie pairs.
    JobConf conf2 = new JobConf(MovieTopDriver.class);
    conf2.setMapperClass(MoviePairsCoOccurMapper.class);
    conf2.setReducerClass(MoviePairsCoOccurReducer.class);

    conf2.setJarByClass(MovieTopDriver.class);

    FileInputFormat.addInputPath(conf2, new Path("temp"));
    FileOutputFormat.setOutputPath(conf2, new Path("freq_temp"));

    conf2.setInputFormat(KeyValueTextInputFormat.class);

    conf2.setMapOutputKeyClass(Text.class);
    conf2.setMapOutputValueClass(IntWritable.class);

    conf2.setOutputKeyClass(Text.class);
    conf2.setOutputValueClass(IntWritable.class);

    // Find top frequent movies along with their names.
    // Output Freq, moviePair
    // Keep a count and output only 20.

    JobConf conf3 = new JobConf(MovieTopDriver.class);
    conf3.setMapperClass(ValueKeyMapper.class);
    conf3.setReducerClass(ValueKeyReducer.class);

    conf3.setJarByClass(MovieTopDriver.class);

    FileInputFormat.addInputPath(conf3, new Path("freq_temp"));
    FileOutputFormat.setOutputPath(conf3, new Path("freq_temp2"));

    conf3.setInputFormat(KeyValueTextInputFormat.class);
    conf3.setMapOutputKeyClass(IntWritable.class);
    conf3.setMapOutputValueClass(Text.class);

    conf3.setOutputKeyClass(IntWritable.class);
    conf3.setOutputValueClass(Text.class);

    // Use only one reducer as we want to sort.
    conf3.setNumReduceTasks(1);

    // To sort in decreasing order.
    conf3.setOutputKeyComparatorClass(LongWritable.DecreasingComparator.class);

    // Find top movie name
    // Use a mapper side join to output names.

    JobConf conf4 = new JobConf(MovieTopDriver.class);
    conf4.setMapperClass(MovieNamesMapper.class);
    conf4.setJarByClass(MovieTopDriver.class);

    FileInputFormat.addInputPath(conf4, new Path("freq_temp2"));
    FileOutputFormat.setOutputPath(conf4, new Path(args[1]));

    conf4.setInputFormat(KeyValueTextInputFormat.class);
    conf4.setMapOutputKeyClass(Text.class);
    conf4.setMapOutputValueClass(Text.class);

    // Run the jobs

    Job job1 = new Job(conf1);
    Job job2 = new Job(conf2);
    Job job3 = new Job(conf3);
    Job job4 = new Job(conf4);

    JobControl jobControl = new JobControl("jobControl");
    jobControl.addJob(job1);
    jobControl.addJob(job2);
    jobControl.addJob(job3);
    jobControl.addJob(job4);
    job2.addDependingJob(job1);
    job3.addDependingJob(job2);
    job4.addDependingJob(job3);
    handleRun(jobControl);

    FileSystem.get(conf2).deleteOnExit(new Path("temp"));
    FileSystem.get(conf3).deleteOnExit(new Path("freq_temp"));
    FileSystem.get(conf4).deleteOnExit(new Path("freq_temp2"));

    System.out.println("Program complete.");
    return 0;
}

更新：我使用的是 Hadoop 1.2.1，我只能使用它，因为我在学校使用集群。

更新：使用了配置而不是设置，但它仍然没有被调用。

public void configure(JobConf jobConf) {

    System.out.println("Setting up system..");

    Path[] cacheFilesLocal;
    try {
        cacheFilesLocal = DistributedCache.getLocalCacheFiles(jobConf);

        for (Path eachPath : cacheFilesLocal) {
            if (eachPath.getName().toString().trim().equals("u.item")) {

                loadMovieNamesHashMap(eachPath);

            }
        }
    } catch (IOException e) {
        e.printStackTrace();
    }

}

在运行方法中添加了以下内容。

DistributedCache.addFileToClassPath(new Path("moviedata"), conf4);
conf4.set("mapred.job.tracker", "local");

Answer 1

您将不得不使用配置方法：

public void configure(JobConf job) {

   }

设置未在Documentation 中定义

Answer 2

如果您的 IDE 支持它，请让您的 IDE 覆盖超类中的方法（在 Eclipse 中它是 Source -> Override/Implement 方法），以查看 IDE 是否认为您的类型（上下文）错误。如果你弄错了，那么 Eclipse 会让你覆盖该方法，插入一个带有正确签名的存根。

确切地说，您需要确定是使用 mapred（旧）还是 map reduce（新）包。您似乎正在使用 mapred 包（注意 Context 是从错误的包中导入的）。如果要使用 mapred 包，请使用 configure() 方法，否则使用 setup() 用于 mapreduce 包

Answer 3

--- 替代解决方案 ---

我仍然无法弄清楚。似乎在 mapper 开始时调用 setup 方法的模型，在任何 map 调用之前，可能只是新 API 的一部分（mapred vs mapreduce）。

我想对只有一个变量差异的多个映射器使用相同的映射方法。 无法覆盖变量，所以我在 map 方法的开头调用了 pulic void setup()，在子映射器中覆盖了它。当然，每次映射调用都会调用它（例如，这些映射器的输入文件中的每一行），但这是我目前效率最低的一个。

public static class Mapper1
    extends MapReduceBase
    implements Mapper<LongWritable, Text, Text, Text>
{
    protected int someVar;

    public void setup()
    {
        System.out.println("[LOG] setup called");
        someVar = 1;
    }

    public void map(
        LongWritable key,
        Text value,
        OutputCollector<Text, Text> output,
        Reporter reporter
    ) throws IOException
    {
        setup();
        System.out.println("someVar: " + String.valueOf(someVar));
        //...
        output.collect(someKey, someValue);
    }
}

public static class Mapper3
    extends Mapper1
{
    //protected int someVar;
    //private int someVar;

    /*
    @Override
    public void setup(Context context)
        throws IOException, InterruptedException
    {
        System.out.println("[LOG] setup called");
        someVar = 2;
    }
    @Override
    public void configure(JobConf jobConf)
    {
        System.out.println("[LOG] configure called");
        someVar = 2;
    }
    */
    @Override
    public void setup()
    {
        System.out.println("[LOG] setup called");
        someVar = 2;
    }
}

Answer 4

我有一个在 Hadoop 1.2.1 上运行的代码（也在 2.2.0 上测试过），它广泛使用设置。这是我的代码中的样子：

    @Override
    public void setup(Context context) throws IllegalArgumentException, IOException {
        logger.debug("setup has been called");
    }

我看到的区别是使用“public”而不是“protected”，并且还使用了@Override，它可以帮助您确定您是否没有正确覆盖该方法。另请注意，我使用的是新 API (org.apache.hadoop.mapreduce)。