import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.io.IOException;
import java.net.URI;
import java.util.*;
import java.util.Arrays;
import java.util.Random;
import java.util.Iterator;
import java.util.StringTokenizer;
import java.util.ArrayList;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
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.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.io.DefaultStringifier;
public class append_views_videoid_combined{
public static class Mapper1 extends Mapper<Object, Text, Text, Text>
{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public static int num;
public Path pathwaysFiles[]=null;
public int n = 2;
private Map<String,String> some_dict = new HashMap<String,String>();
public void setup(Context context) {
Configuration conf = context.getConfiguration();
try{
pathwaysFiles = DistributedCache.getLocalCacheFiles(conf);
for (Path path: pathwaysFiles){
String name = path.getName();
if (name.compareTo("some_dict.txt") == 0) {
BufferedReader fis = new BufferedReader(new InputStreamReader (new FileInputStream(path.toString())));
String line = "";
while ((line = fis.readLine()) != null){
line.replaceAll("\n","");
String str = line;
some_dict.put(str,"1");
}
} else {
continue;
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line,"\n");
String temp = null;
String Did = null;
while (tokenizer.hasMoreTokens()) {
temp = tokenizer.nextToken();
String[] info = temp.split(",");
String timestamp = info[0].trim();
// String days = info[0].substring(0,10);
String userid = info[1].trim();
String webid = info[2].substring(0,2);
String device = info[3].trim();
String videoid = info[4].trim();
String videoid_alter = info[5].trim();
String val = top_views_dict.get(videoid);
String top1k = "";
if ( val != null ) {
top1k = "1";
}else{
top1k = "0";
}
String user = userid + device;
String output_str = timestamp + ',' + user + ',' + webid + ',' + top1k;
context.write( new Text(videoid_alter), new Text(output_str) );
}
}
}
public static class Mapper2 extends Mapper<Object, Text, Text, Text>
{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line,"\n");
String temp = null;
String Did = null;
while (tokenizer.hasMoreTokens()) {
temp = tokenizer.nextToken();
String[] info = temp.split("\t");
String videoid = info[0].trim();
String categoryid = info[1].trim();
context.write( new Text(videoid) , new Text( "X" + categoryid ) );
}
}
}
public static class StatsReducer extends Reducer<Text,Text,Text,Text>
{
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException
{
int mark = 0;
String categoryid = "";
ArrayList<String> A1 = new ArrayList<String>();
try{
for (Text val : values)
{
String temp = val.toString();
if (temp.substring(0,1).compareTo("X") == 0)
{
mark = 1;
categoryid = temp.substring(1, temp.length());
}
else
{
A1.add(temp);
}
}
if (mark == 1)
{
for (String str : A1)
{
context.write(key, new Text(str + "," + categoryid));
}
}
else
{
for (String str : A1)
{
context.write(key, new Text(str + "," + "0"));
}
}
}catch(Throwable e){}
}
}
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
DistributedCache.addCacheFile(new URI(args[3]), conf);
Job job = new Job(conf, "append_views_videoid_combined");
job.setJarByClass(append_views_videoid_combined.class);
job.setReducerClass(StatsReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setNumReduceTasks(1);
MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, Mapper1.class);
MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, Mapper2.class);
FileOutputFormat.setOutputPath(job, new Path(args[2]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}