hadoop - 有什么建议可以同时将两个不同的数据集读入 Hadoop？

Question

亲爱的hadooper:
我是hadoop的新手，最近尝试实现一个算法。

该算法需要计算一个矩阵，该矩阵表示每两对歌曲的不同评分。我已经这样做了，输出是一个 600000*600000 稀疏矩阵，我存储在我的 HDFS 中。我们称这个数据集为 A(大小=160G)

现在，我需要阅读用户的个人资料来预测他们对特定歌曲的评分。所以我需要先读取用户的个人资料(5G大小)，让我们调用这个数据集B，然后使用数据集A进行计算。

但现在我不知道如何从一个 hadoop 程序中读取这两个数据集。或者我可以将数据集 B 读入 RAM 然后进行计算吗？(我想我不能，因为 HDFS 是一个分布式系统，我无法将数据集 B 读入单个机器的内存)。

有什么建议么？

Answer 1

您可以使用两个 Map 函数，每个 Map 函数可以处理一个数据集，如果您想实现不同的处理。您需要在您的工作配置中注册您的 map 。例如:

           public static class FullOuterJoinStdDetMapper extends MapReduceBase implements Mapper <LongWritable ,Text ,Text, Text>
    {
            private String person_name, book_title,file_tag="person_book#";
            private String emit_value = new String();
            //emit_value = "";
            public void map(LongWritable key, Text values, OutputCollector<Text,Text>output, Reporter reporter)
                     throws IOException
            {
                    String line = values.toString();
                    try
                    {
                            String[] person_detail = line.split(",");
                            person_name = person_detail[0].trim();
                            book_title = person_detail[1].trim();
                    }
                    catch (ArrayIndexOutOfBoundsException e)
                    {
                            person_name = "student name missing";
                     }
                    emit_value = file_tag + person_name;
                    output.collect(new Text(book_title), new Text(emit_value));
            }

    }


       public static class FullOuterJoinResultDetMapper extends MapReduceBase implements  Mapper <LongWritable ,Text ,Text, Text>
     {
            private String author_name, book_title,file_tag="auth_book#";
            private String emit_value = new String();

//发射值 = "";
公共(public)无效映射(LongWritable 键、文本值、OutputCollector 输出、Reporter 记者)
抛出 IOException
{
字符串行 = values.toString();
尝试
{
String[] author_detail = line.split(",");
author_name = author_detail[1].trim();
book_title = author_detail[0].trim();
}
捕捉(ArrayIndexOutOfBoundsException e)
{
author_name = "未出现在考试中";
}

                          emit_value = file_tag + author_name;                                     
                         output.collect(new Text(book_title), new Text(emit_value));
                 }

             }


       public static void main(String args[])
                    throws Exception
    {

           if(args.length !=3)
                    {
                            System.out.println("Input outpur file missing");
                            System.exit(-1);
                    }


            Configuration conf = new Configuration();
            String [] argum = new GenericOptionsParser(conf,args).getRemainingArgs();
            conf.set("mapred.textoutputformat.separator", ",");
            JobConf mrjob = new JobConf();
            mrjob.setJobName("Inner_Join");
            mrjob.setJarByClass(FullOuterJoin.class);

            MultipleInputs.addInputPath(mrjob,new Path(argum[0]),TextInputFormat.class,FullOuterJoinStdDetMapper.class);
            MultipleInputs.addInputPath(mrjob,new Path(argum[1]),TextInputFormat.class,FullOuterJoinResultDetMapper.class);

            FileOutputFormat.setOutputPath(mrjob,new Path(args[2]));
            mrjob.setReducerClass(FullOuterJoinReducer.class);

            mrjob.setOutputKeyClass(Text.class);
            mrjob.setOutputValueClass(Text.class);

            JobClient.runJob(mrjob);
    }