I'm learning some MapReduce but I'm running into some problems, here's the situation: i've two files: "users" contains a list of users with some of their data (gender,age,country,etc...) the file looks like this:
user_000003 m 22 United States Oct 30, 2005
"songs" contains data of songs listened by all the users (userid, date and time of listening,artist id, artist name, song id, song title):
user_000999 2008-12-11T22:52:33Z b7ffd2af-418f-4be2-bdd1-22f8b48613da Nine Inch Nails 1d1bb32a-5bc6-4b6f-88cc-c043f6c52509 7 Ghosts I
the goal is to find the k most listened song in certain countries. with k and a list of countries provided in input.
I decided to use the MultipleInputs class for the mapper so one mapper will output a set of key-value pairs that looks like this: . the other mapper will output . as far as I know i should be able to read all the values paired with a certain key (so I should find the country and a certain number of songs in the list of values associated with a userID) in the reducer and output a set of file with pairs to be red by another MapReduce job.
I'm pretty sure the mappers do their job as I was able to write their output with the reducer.
UPDATE: the files are passed to the mappers with the following code:
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);
Mappers code:
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()
}
Combiner code:
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));
}
}
}
}
The problems comes when I try to output the pairs with the reducer:
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException
{
for (Text value : values)
{
context.write(key,value);
}
}
}
I used " ||| " to build the artist+title string the problem is that country remains "foo". I think I should see at least one row of the output with the correct nation as the key but the output it's always "foo" (2,5kb songs file):
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
what am I doing wrong?
PS I should be able to avoid the second job if I use a custom combiner, does the combiner act exactly like a reducer?
