java - 用于散列的 Hadoop Map Reduce 程序

Question

我在Hadoop中编写了一个Map Reduce程序，用于对文件的所有记录进行哈希处理，并将哈希值作为附加属性附加到每条记录，然后输出到Hadoop文件系统这是我写的代码

public class HashByMapReduce
{
public static class LineMapper extends Mapper<Text, Text, Text, Text>
{
    private Text word = new Text();

    public void map(Text key, Text value, Context context) throws IOException,    InterruptedException
      {
        key.set("single")
        String line = value.toString();
            word.set(line);
            context.write(key, line);

    }
}
public static class LineReducer
extends Reducer<Text,Text,Text,Text>
{
    private Text result = new Text();
    public void reduce(Text key, Iterable<Text> values,
    Context context
    ) throws IOException, InterruptedException
    {
        String translations = "";
        for (Text val : values)
        {
            translations = val.toString()+","+String.valueOf(hash64(val.toString())); //Point of Error 

        result.set(translations);
        context.write(key, result);
        }
    }
}
public static void main(String[] args) throws Exception
{
    Configuration conf = new Configuration();
    Job job = new Job(conf, "Hashing");
    job.setJarByClass(HashByMapReduce.class);
    job.setMapperClass(LineMapper.class);
    job.setReducerClass(LineReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    job.setInputFormatClass(KeyValueTextInputFormat.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
}

我编写这段代码的逻辑是，每一行都由 Map 方法读取，该方法将所有值分配给一个键，然后传递给相同的 Reducer 方法。它将每个值传递给 hash64() 函数。

但我看到它向散列函数传递了一个空值(空值)。我不是无法弄清楚为什么？提前致谢

Answer 1

问题的原因很可能是由于使用了 KeyValueTextInputFormat。来自 Yahoo tutorial :

  InputFormat:          Description:       Key:                     Value:

  TextInputFormat       Default format;    The byte offset          The line contents 
                        reads lines of     of the line                            
                        text files

  KeyValueInputFormat   Parses lines       Everything up to the     The remainder of                      
                        into key,          first tab character      the line
                        val pairs 

它打破了你的输入行 wrt tab 字符。我想您的行中没有 tab 。结果 LineMapper 中的 key 是整行，而没有任何内容作为 value 传递(不确定 null 或为空)。

根据你的代码，我认为你应该更好地使用 TextInputFormat 类作为你的输入格式，它将行偏移量作为 key 并将完整的行作为 value .这应该可以解决您的问题。

编辑:我运行您的代码并进行以下更改，它似乎工作正常:

1. 将输入格式更改为 TextInputFormat 并相应地更改映射器的声明
2. 向 job 添加了正确的 setMapOutputKeyClass 和 setMapOutputValueClass。这些不是强制性的，但通常会在运行时产生问题。
3. 删除了您的 ket.set("single") 并向 Mapper 添加了一个私有(private)外键。
4. 由于您没有提供hash64 方法的详细信息，因此我使用String.toUpperCase 进行测试。

如果问题仍然存在，那么我确定您的哈希方法没有很好地处理 null。

完整代码:

 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.mapreduce.Job;
 import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.Reducer;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 public class HashByMapReduce {
 public static class LineMapper extends
        Mapper<LongWritable, Text, Text, Text> {
    private Text word = new Text();
    private Text outKey = new Text("single");

    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        word.set(line);
        context.write(outKey, word);
    }
}

public static class LineReducer extends Reducer<Text, Text, Text, Text> {
    private Text result = new Text();

    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        String translations = "";
        for (Text val : values) {
            translations = val.toString() + ","
                    + val.toString().toUpperCase(); // Point of Error

            result.set(translations);
            context.write(key, result);
        }
    }
}

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = new Job(conf, "Hashing");
    job.setJarByClass(HashByMapReduce.class);
    job.setMapperClass(LineMapper.class);
    job.setReducerClass(LineReducer.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    job.setInputFormatClass(TextInputFormat.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
}