hadoop-mapreduce reducer-combiner input

Question

我正在学习一些MapReduce，但是我遇到了一些问题，情况是这样的:我有两个文件:“users”包含一个用户列表，其中包含一些用户数据(性别、年龄、国家等)...)文件看起来像这样:

user_000003  m  22  United States   Oct 30, 2005

“歌曲”包含所有用户收听的歌曲的数据（用户ID，收听日期和时间，艺术家ID，艺术家姓名，歌曲ID，歌曲标题）：

user_000999 2008-12-11T22:52:33Z    b7ffd2af-418f-4be2-bdd1-22f8b48613da    Nine Inch Nails 1d1bb32a-5bc6-4b6f-88cc-c043f6c52509    7 Ghosts I

目标是在某些国家找到k首最受欢迎的歌曲。k和输入中提供的国家列表。

我决定为映射器使用MultipleInputs类，这样一个映射器将输出一组如下所示的键值对:。另一个映射器将输出。据我所知，我应该能够在reducer中读取与某个键配对的所有值(因此我应该在与userID相关联的值列表中找到国家和一定数量的歌曲),并输出一组由另一个MapReduce作业读取的文件。

我很确定映射程序完成了他们的工作，因为我能够用reducer编写他们的输出。

更新：使用以下代码将文件传递给映射器：

Job job = Job.getInstance(conf);

        MultipleInputs.addInputPath(job, new Path(songsFile), TextInputFormat.class, SongMapper.class);
        MultipleInputs.addInputPath(job, new Path(usersFile), TextInputFormat.class, UserMapper.class);

        FileOutputFormat.setOutputPath(job, new Path(outFile));
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

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

        job.setJarByClass(Songs.class);

        job.setCombinerClass(Combiner.class);
        job.setReducerClass(UserSongReducer.class);

映射器代码：

public static class UserMapper extends Mapper<LongWritable, Text, Text, Text>{

        //empty cleanUp()

        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            String record = value.toString();               
            String[] userData = record.split("\t");
            if(userData.length>3 && !userData[3].equals(""))
            {
                context.write(new Text(userData[0]), new Text(userData[3]));
            }        


        }

        //empty run() and setup()

    }

    public static class SongMapper extends Mapper<LongWritable, Text, Text, Text>{

        //empty cleanUp()

        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            String record = value.toString();               
            String[] songData = record.split("\t");
            if(songData.length>3 && !songData[3].equals(""))
            {
                context.write(new Text(songData[0]), new Text(songData[5]+" |||  "+songData[3]));
            }        


        }

        //empty run() and setup()

    }

组合器代码：

public static class Combiner extends Reducer<Text, Text, Text, Text> 
    {

        private boolean isCountryAllowed(String toCheck, String[] countries)
        {

            for(int i=0; i<countries.length;i++)
            {
                if(toCheck.equals(countries[i]))
                    return true;
            }
            return false;
        }

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException 
        {



            ArrayList<String> list = new ArrayList<String>();



            String country = "foo";
            for(Text value : values) 
            {
                if(!value.toString().contains(" ||| "))
                {
                    country = value.toString();
                }else
                {
                    list.add(value.toString());
                }

            }

            if(isCountryAllowed(country, context.getConfiguration().getStrings("countries")))
            {

                for (String listVal : list) 
                {
                    context.write(new Text(country), 
                            new Text(listVal));
                }
            }

         }
    }

当我试图用减速器输出线对时，问题出现了:

public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException 
        {


             for (Text value : values) 
            {
                context.write(key,value);
            }
            }

         }

我使用“||”来构建艺术家标题字符串，问题是country仍然是“foo”。我想我应该看到至少一行输出，正确的国家作为关键，但输出总是“foo”(2，5kb歌曲文件):

foo Deep Dish |||  Fuck Me Im Famous (Pacha Ibiza)-09-28-2007
foo Vnv Nation |||  Kingdom
foo Les Fleur De Lys |||  Circles
foo Home Video |||  Penguin
foo Of Montreal |||  Will You Come And Fetch Me
foo Godspeed You! Black Emperor |||  Bbf3
foo Alarum |||  Sustained Connection
foo Sneaker Pimps |||  Walking Zero
foo Cecilio And Kapono |||  I Love You
foo Garbage |||  Vow
foo The Brian Setzer Orchestra |||  Gettin' In The Mood
foo Nitin Sawhney |||  Sunset (J-Walk Remix)
foo Nine Inch Nails |||  Heresy
foo Collective Soul |||  Crowded Head
foo Vicarious Bliss |||  Limousine
foo Noisettes |||  Malice In Wonderland
foo Black Rebel Motorcycle Club |||  Lien On Your Dreams
foo Mae |||  Brink Of Disaster
foo Michael Andrews |||  Rosie Darko
foo A Perfect Circle |||  Blue

我做错了什么？

PS如果我使用自定义组合器，我应该可以避免第二个工作，组合器的作用是否完全像减速器？

Answer 1

从你的代码中我看到，国家应该是“foo”，你到底想实现什么？

 // IN YOUR MAPPER THE VALUE IS WRITTEN USING |||

context.write(new Text(songData[0]), new Text(songData[5]+" |||  "+songData[3]));

// THEN VALUE WILL ALWAYS CONTAIN ||| AND WILL NEVER CHANGE THE COUNTRY THAT WAS SET TO TRUE
 String country = "foo";
     if(!value.toString().contains(" ||| ")) //never --- 

然后输出国家变量：

 context.write(new Text(country),