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.text.ParseException;
import java.net.URI;
import java.util.*;
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.SequenceFileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.io.DefaultStringifier;
public class refresh_switch_web_ornot{
public static class StatsMapper extends Mapper<Object, Text, Text, IntWritable>
{
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[] info = line.split("\t");
String id = info[0];
String caption = info[1];
some_dict.put(original_videoid, altered_videoid);
}
} else {
continue;
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
Configuration conf = context.getConfiguration();
String threshold_s = conf.get("threshold");
long threshold = Long.parseLong(threshold_s);
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line,"\n");
String temp = null;
String Did = null;
while (tokenizer.hasMoreTokens())
{
// try {
temp = tokenizer.nextToken();
temp = temp.trim();
String[] info = temp.split("\t");
String userid = info[0];
String[] others = info[1].split(";");
// Initialize
int flag = 0; // first loop flag
int flag_2 = 0; // Second loop flag
String webid = "";
String videoid = "";
String categoryid = "";
String top1k_or_not = "";
// IRQ flag
int irq_flag = 0;
// Count variables
int cross_top = 0; // Count of cross to a top video of other website
int cross_sum = 0; // Count of all cross in 2 hours
// Final Output: cross_top/cross_sum
// Temp arg
String previous_previous_videoid = "";
String previous_webid = "";
String previous_videoid = "";
String previous_categoryid = "";
String previous_top1k_or_not = "";
// Time
java.text.SimpleDateFormat format = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
java.util.Date previous_timestamp = new java.util.Date(2011,10,28,10,12,50);
java.util.Date timestamp = new java.util.Date(2011,10,28,10,12,50);
try{
// 0092148ee1191eebfb7576d4fb8b5419fca647,Android4.1.2 X;2014-12-24 11:47:48,26,0;2014-12-24 11:50:37,26,0;2014-12-24 11:52:57,26,0;2014-12-24 11:56:24,26,0;2014-12-24 12:01:57,26,0
previous_timestamp = format.parse("2007-12-24 00:00:00");
timestamp = format.parse("2007-12-24 00:00:00");
}catch(ParseException e){e.printStackTrace();}
long stop = 100000;
long previous_stop = 0;
int in_session_flag = 0;
int count = 1;
String start_webid = "";
String target_videoid = "";
for (String str : others)
{
if (flag == 0) {
flag = 1; // first loop, except: X
// continue;
} else {
// Read First Data
if (flag_2 == 0) {
irq_flag = 0;
flag_2 = 1;
// Load this one
String[] temp_1 = str.split(",");
try{
timestamp = format.parse(temp_1[0].trim());
}catch(ParseException e){e.printStackTrace();}
videoid = temp_1[1].trim();
// Change to same videoid
String val = same_videoid_dict.get(videoid);
if ( val!=null) {
videoid = val;
}
webid = temp_1[2].trim();
categoryid = temp_1[3].trim();
top1k_or_not = temp_1[4].trim();
} else { // Normal Situation
irq_flag = 0;
// Load the one before last one
previous_previous_videoid = previous_videoid;
// Load the last one
previous_timestamp = timestamp;
previous_webid = webid;
previous_videoid = videoid;
previous_categoryid = categoryid;
previous_top1k_or_not = top1k_or_not;
previous_stop = stop;
// Load this one
String[] temp_1 = str.split(",");
try{
timestamp = format.parse(temp_1[0].trim());
}catch(ParseException e){e.printStackTrace();}
videoid = temp_1[1].trim();
String val = some_dict.get(videoid);
if ( val!=null) {
videoid = val;
}
webid = temp_1[2].trim();
categoryid = temp_1[3].trim();
top1k_or_not = temp_1[4].trim();
stop = timestamp.getTime() - previous_timestamp.getTime();
if (stop <= threshold*1000) {
if ( !webid.equals(previous_webid) ) { // Switch to a different web
if ( previous_stop <= threshold*1000 ) {
if ( previous_previous_videoid.equals(previous_videoid) ) { // Refresh in former website
context.write(new Text("Switch after refresh count"), one);
context.write(new Text("x2y_"+previous_webid+"_"+webid), one);
}
}
}
}
}
}
}
// }
// catch (ParseException e) {
// //Handle exception here, most of the time you will just log it.
// e.printStackTrace();
// }
}
}
}
public static class StatsReducer extends Reducer<Text,IntWritable,Text,IntWritable>
{
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
{
int sum = 0;
for (IntWritable val : values)
{
sum += val.get();
}
context.write(key, new IntWritable(sum));
}
}
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
DistributedCache.addCacheFile(new URI(args[2]), conf);
conf.set("threshold",args[3]);
Job job = new Job(conf, "refresh_switch_web_ornot");
job.setJarByClass(refresh_switch_web_ornot.class);
job.setMapperClass(StatsMapper.class);
job.setReducerClass(StatsReducer.class);
// job.setMapOutputKeyClass(Text.class);
// job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setNumReduceTasks(1);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}