You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2013/04/16 01:13:10 UTC
svn commit: r1468268 [2/3] - in /pig/trunk: ./
test/e2e/pig/udfs/java/org/apache/pig/test/udf/storefunc/ test/perf/
test/perf/pigmix/ test/perf/pigmix/bin/ test/perf/pigmix/conf/
test/perf/pigmix/src/ test/perf/pigmix/src/java/ test/perf/pigmix/src/jav...
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L12.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L12.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L12.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L12.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,262 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L12 {
+
+ public static class HighestValuePagePerUser extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, DoubleWritable>,
+ Reducer<Text, DoubleWritable, Text, DoubleWritable> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, DoubleWritable> oc,
+ Reporter reporter) throws IOException {
+
+ List<Text> fields = Library.splitLine(val, '');
+
+ // Filter out null users and query terms.
+ if (fields.get(0).getLength() == 0 &&
+ fields.get(3).getLength() == 0) return;
+ try {
+ oc.collect(fields.get(0),
+ new DoubleWritable(Double.valueOf(fields.get(6).toString())));
+ } catch (NumberFormatException nfe) {
+ oc.collect(fields.get(0), new DoubleWritable(0));
+ }
+ }
+
+ public void reduce(
+ Text key,
+ Iterator<DoubleWritable> iter,
+ OutputCollector<Text, DoubleWritable> oc,
+ Reporter reporter) throws IOException {
+ double max = Double.MIN_VALUE;
+
+ while (iter.hasNext()) {
+ double d = iter.next().get();
+ max = max > d ? max : d;
+ }
+
+ oc.collect(key, new DoubleWritable(max));
+ }
+ }
+
+ public static class TotalTimespentPerTerm extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, LongWritable>,
+ Reducer<Text, LongWritable, Text, LongWritable> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, LongWritable> oc,
+ Reporter reporter) throws IOException {
+ List<Text> fields = Library.splitLine(val, '');
+
+ // Filter out non-null users
+ if (fields.get(0).getLength() != 0) return;
+ try {
+ oc.collect(fields.get(3),
+ new LongWritable(Long.valueOf(fields.get(2).toString())));
+ } catch (NumberFormatException nfe) {
+ oc.collect(fields.get(3), new LongWritable(0));
+ }
+
+ }
+
+ public void reduce(
+ Text key,
+ Iterator<LongWritable> iter,
+ OutputCollector<Text, LongWritable> oc,
+ Reporter reporter) throws IOException {
+ long sum = 0;
+
+ while (iter.hasNext()) sum += iter.next().get();
+ oc.collect(key, new LongWritable(sum));
+ }
+ }
+
+ public static class QueriesPerAction extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, LongWritable>,
+ Reducer<Text, LongWritable, Text, LongWritable> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, LongWritable> oc,
+ Reporter reporter) throws IOException {
+ List<Text> fields = Library.splitLine(val, '');
+
+ // Filter out non-null users and non-null queries
+ if (fields.get(0).getLength() != 0 ||
+ fields.get(3).getLength() != 0) return;
+ oc.collect(fields.get(1), new LongWritable(1));
+
+ }
+
+ public void reduce(
+ Text key,
+ Iterator<LongWritable> iter,
+ OutputCollector<Text, LongWritable> oc,
+ Reporter reporter) throws IOException {
+ long cnt = 0;
+
+ while (iter.hasNext()) {
+ iter.next();
+ cnt++;
+ }
+ oc.collect(key, new LongWritable(cnt));
+ }
+ }
+
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ String user = System.getProperty("user.name");
+ JobConf lp = new JobConf(L12.class);
+ lp.setJobName("L12 Find Highest Value Page Per User");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(DoubleWritable.class);
+ lp.setMapperClass(HighestValuePagePerUser.class);
+ lp.setCombinerClass(HighestValuePagePerUser.class);
+ lp.setReducerClass(HighestValuePagePerUser.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/highest_value_page_per_user"));
+ lp.setNumReduceTasks(Integer.parseInt(parallel));
+ Job loadPages = new Job(lp);
+
+ JobConf lu = new JobConf(L12.class);
+ lu.setJobName("L12 Find Total Timespent per Term");
+ lu.setInputFormat(TextInputFormat.class);
+ lu.setOutputKeyClass(Text.class);
+ lu.setOutputValueClass(LongWritable.class);
+ lu.setMapperClass(TotalTimespentPerTerm.class);
+ lu.setCombinerClass(TotalTimespentPerTerm.class);
+ lu.setReducerClass(TotalTimespentPerTerm.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lu.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lu, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lu, new Path(outputDir + "/total_timespent_per_term"));
+ lu.setNumReduceTasks(Integer.parseInt(parallel));
+ Job loadUsers = new Job(lu);
+
+ JobConf join = new JobConf(L12.class);
+ join.setJobName("L12 Find Queries Per Action");
+ join.setInputFormat(TextInputFormat.class);
+ join.setOutputKeyClass(Text.class);
+ join.setOutputValueClass(LongWritable.class);
+ join.setMapperClass(QueriesPerAction.class);
+ join.setCombinerClass(QueriesPerAction.class);
+ join.setReducerClass(QueriesPerAction.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ join.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(join, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(join, new Path(outputDir + "/queries_per_action"));
+ join.setNumReduceTasks(Integer.parseInt(parallel));
+ Job joinJob = new Job(join);
+
+ JobControl jc = new JobControl("L12 join");
+ jc.addJob(loadPages);
+ jc.addJob(loadUsers);
+ jc.addJob(joinJob);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L13.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L13.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L13.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L13.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,221 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+public class L13 {
+
+ public static class ReadLeftPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ List<Text> fields = Library.splitLine(val, '');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ oc.collect(fields.get(0), new Text("1" + fields.get(6)));
+ }
+ }
+
+ public static class ReadRightPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ // Prepend an index to the value so we know which file
+ // it came from.
+ List<Text> fields = Library.splitLine(val, '\u0001');
+ if (fields.size()==0) return;
+ oc.collect(fields.get(0), new Text("2"+fields.get(1)));
+
+ }
+ }
+
+ public static class Join extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ // For each value, figure out which file it's from and store it
+ // accordingly.
+ List<String> first = new ArrayList<String>();
+ List<String> second = new ArrayList<String>();
+
+ while (iter.hasNext()) {
+ Text txt = iter.next();
+ if (txt.charAt(0) == '1') {
+ first.add(txt.toString().substring(1));
+ }
+ else second.add(txt.toString().substring(1));
+ reporter.setStatus("OK");
+ }
+
+ reporter.setStatus("OK");
+
+ if (first.size() == 0) return;
+ if (second.size() == 0) second.add(null);
+
+ // Do the cross product
+ for (String s1 : first) {
+ for (String s2 : second) {
+ if (s2==null) oc.collect(null, new Text(key + "\t" + s1 + "\t\t"));
+ else oc.collect(null, new Text(key + "\t" + s1 + "\t" + key + "\t" + s2));
+ }
+ }
+ first.clear();
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ String user = System.getProperty("user.name");
+ JobConf lp = new JobConf(L13.class);
+ lp.setJobName("L13 Load Left Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadLeftPageViews.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/indexed_left_pages"));
+ lp.setNumReduceTasks(0);
+ Job loadPages = new Job(lp);
+
+ JobConf lu = new JobConf(L13.class);
+ lu.setJobName("L13 Load Right Page Views");
+ lu.setInputFormat(TextInputFormat.class);
+ lu.setOutputKeyClass(Text.class);
+ lu.setOutputValueClass(Text.class);
+ lu.setMapperClass(ReadRightPageViews.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lu.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lu, new Path(inputDir + "/power_users_samples"));
+ FileOutputFormat.setOutputPath(lu, new Path(outputDir + "/indexed_right_pages"));
+ lu.setNumReduceTasks(0);
+ Job loadUsers = new Job(lu);
+
+ JobConf join = new JobConf(L13.class);
+ join.setJobName("L13 Join Two Pages");
+ join.setInputFormat(KeyValueTextInputFormat.class);
+ join.setOutputKeyClass(Text.class);
+ join.setOutputValueClass(Text.class);
+ join.setMapperClass(IdentityMapper.class);
+ join.setReducerClass(Join.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ join.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_left_pages"));
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_right_pages"));
+ FileOutputFormat.setOutputPath(join, new Path(outputDir + "/L13out"));
+ join.setNumReduceTasks(Integer.parseInt(parallel));
+ Job joinJob = new Job(join);
+ joinJob.addDependingJob(loadPages);
+ joinJob.addDependingJob(loadUsers);
+
+ JobControl jc = new JobControl("L13 join");
+ jc.addJob(loadPages);
+ jc.addJob(loadUsers);
+ jc.addJob(joinJob);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L14.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L14.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L14.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L14.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,223 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L14 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ List<Text> fields = Library.splitLine(val, '');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ Text outVal = new Text("1" + fields.get(6).toString());
+ oc.collect(fields.get(0), outVal);
+ }
+ }
+
+ public static class ReadUsers extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ List<Text> fields = Library.splitLine(val, '\u0001');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ Text outVal = new Text("2");
+ oc.collect(fields.get(0), outVal);
+
+ }
+ }
+
+ public static class Join extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ // For each value, figure out which file it's from and store it
+ // accordingly.
+ List<String> first = new ArrayList<String>();
+ List<String> second = new ArrayList<String>();
+
+ while (iter.hasNext()) {
+ Text t = iter.next();
+ String value = t.toString();
+ if (value.charAt(0) == '1') first.add(value.substring(1));
+ else second.add(value.substring(1));
+ reporter.setStatus("OK");
+ }
+
+ reporter.setStatus("OK");
+
+ if (first.size() == 0 || second.size() == 0) return;
+
+ // Do the cross product, and calculate the sum
+ for (String s1 : first) {
+ for (String s2 : second) {
+ try {
+ oc.collect(null, new Text(key.toString() + "\t" + s1 + "\t" + key.toString()));
+ } catch (NumberFormatException nfe) {
+ }
+ }
+ }
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ String user = System.getProperty("user.name");
+ JobConf lp = new JobConf(L14.class);
+ lp.setJobName("L14 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views_sorted"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/indexed_pages_14"));
+ lp.setNumReduceTasks(0);
+ Job loadPages = new Job(lp);
+
+ JobConf lu = new JobConf(L14.class);
+ lu.setJobName("L14 Load Users");
+ lu.setInputFormat(TextInputFormat.class);
+ lu.setOutputKeyClass(Text.class);
+ lu.setOutputValueClass(Text.class);
+ lu.setMapperClass(ReadUsers.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lu.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lu, new Path(inputDir + "/users_sorted"));
+ FileOutputFormat.setOutputPath(lu, new Path(outputDir + "/indexed_users_14"));
+ lu.setNumReduceTasks(0);
+ Job loadUsers = new Job(lu);
+
+ JobConf join = new JobConf(L14.class);
+ join.setJobName("L14 Join Users and Pages");
+ join.setInputFormat(KeyValueTextInputFormat.class);
+ join.setOutputKeyClass(Text.class);
+ join.setOutputValueClass(Text.class);
+ join.setMapperClass(IdentityMapper.class);
+ join.setReducerClass(Join.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ join.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_pages_14"));
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_users_14"));
+ FileOutputFormat.setOutputPath(join, new Path(outputDir + "/L14out"));
+ join.setNumReduceTasks(Integer.parseInt(parallel));
+ Job joinJob = new Job(join);
+ joinJob.addDependingJob(loadPages);
+ joinJob.addDependingJob(loadUsers);
+
+ JobControl jc = new JobControl("L14 join");
+ jc.addJob(loadPages);
+ jc.addJob(loadUsers);
+ jc.addJob(joinJob);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L15.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L15.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L15.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L15.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,215 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L15 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ // Split the line
+ List<Text> fields = Library.splitLine(val, '');
+ if (fields.size() != 9) return;
+
+ StringBuffer sb = new StringBuffer();
+ sb.append(fields.get(1).toString()); // action
+ sb.append('');
+ sb.append(fields.get(6).toString()); // estimated_revenue
+ sb.append('');
+ sb.append(fields.get(2).toString()); // timespent
+ sb.append('');
+ oc.collect(fields.get(0), new Text(sb.toString()));
+ }
+ }
+
+ public static class Combiner extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ HashSet<Text> hash1 = new HashSet<Text>();
+ HashSet<Text> hash2 = new HashSet<Text>();
+ HashSet<Text> hash3 = new HashSet<Text>();
+ while (iter.hasNext()) {
+ List<Text> vals = Library.splitLine(iter.next(), '');
+ try {
+ hash1.add(vals.get(0));
+ hash2.add(vals.get(1));
+ hash3.add(vals.get(2));
+ } catch(NumberFormatException nfe) {
+ }
+ }
+ Double rev= new Double(0.0);
+ Integer ts=0;
+ try {
+ for (Text t : hash2) rev += Double.valueOf(t.toString());
+ for (Text t : hash3) ts += Integer.valueOf(t.toString());
+ } catch (NumberFormatException e) {
+ }
+ StringBuffer sb = new StringBuffer();
+ sb.append((new Integer(hash1.size())).toString());
+ sb.append("");
+ sb.append(rev.toString());
+ sb.append("");
+ sb.append(ts.toString());
+ sb.append("");
+ oc.collect(key, new Text(sb.toString()));
+ reporter.setStatus("OK");
+ }
+ }
+ public static class Group extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ HashSet<Text> hash1 = new HashSet<Text>();
+ HashSet<Text> hash2 = new HashSet<Text>();
+ HashSet<Text> hash3 = new HashSet<Text>();
+ while (iter.hasNext()) {
+ List<Text> vals = Library.splitLine(iter.next(), '');
+ try {
+ hash1.add(vals.get(0));
+ hash2.add(vals.get(1));
+ hash3.add(vals.get(2));
+ }catch(NumberFormatException nfe) {
+ }
+ }
+ Integer ts=0;
+ Double rev=new Double(0.0);
+ for (Text t : hash2) rev += Double.valueOf(t.toString());
+ for (Text t : hash3) ts += Integer.valueOf(t.toString());
+ StringBuffer sb = new StringBuffer();
+ sb.append((new Integer(hash1.size())).toString());
+ sb.append("");
+ sb.append(rev.toString());
+ sb.append("");
+ sb.append(ts.toString());
+ oc.collect(key, new Text(sb.toString()));
+ reporter.setStatus("OK");
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ JobConf lp = new JobConf(L15.class);
+ lp.setJobName("L15 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ lp.setCombinerClass(Combiner.class);
+ lp.setReducerClass(Group.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/L15out"));
+ lp.setNumReduceTasks(Integer.parseInt(parallel));
+ Job group = new Job(lp);
+
+ JobControl jc = new JobControl("L15 join");
+ jc.addJob(group);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L16.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L16.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L16.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L16.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,159 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.TreeSet;
+import java.util.Iterator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Properties;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L16 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ // Split the line
+ List<Text> fields = Library.splitLine(val, '');
+ if (fields.size() != 9) return;
+
+ oc.collect(fields.get(0), fields.get(6));
+ }
+ }
+
+ public static class Group extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ Set<Double> hash2 = new TreeSet<Double>();
+ while (iter.hasNext()) {
+ List<Text> vals = Library.splitLine(iter.next(), '');
+ try {
+ hash2.add(Double.valueOf(vals.get(0).toString()));
+ }catch(NumberFormatException nfe) {
+ }
+ }
+ Double rev=new Double(0);
+ for (Double d : hash2) rev += d;
+ oc.collect(key, new Text(rev.toString()));
+ reporter.setStatus("OK");
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ JobConf lp = new JobConf(L16.class);
+ lp.setJobName("L16 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ lp.setReducerClass(Group.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/L16out"));
+ lp.setNumReduceTasks(Integer.parseInt(parallel));
+ Job group = new Job(lp);
+
+ JobControl jc = new JobControl("L16 join");
+ jc.addJob(group);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L17.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L17.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L17.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L17.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,284 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L17 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ List<Text> vals = Library.splitLine(val, '');
+ if (vals.size() != 27) return;
+ StringBuffer key = new StringBuffer();
+ key.append(vals.get(0).toString());
+ key.append("");
+ key.append(vals.get(1).toString());
+ key.append("");
+ key.append(vals.get(2).toString());
+ key.append("");
+ key.append(vals.get(3).toString());
+ key.append("");
+ key.append(vals.get(4).toString());
+ key.append("");
+ key.append(vals.get(5).toString());
+ key.append("");
+ key.append(vals.get(6).toString());
+ key.append("");
+ key.append(vals.get(9).toString());
+ key.append("");
+ key.append(vals.get(10).toString());
+ key.append("");
+ key.append(vals.get(11).toString());
+ key.append("");
+ key.append(vals.get(12).toString());
+ key.append("");
+ key.append(vals.get(13).toString());
+ key.append("");
+ key.append(vals.get(14).toString());
+ key.append("");
+ key.append(vals.get(15).toString());
+ key.append("");
+ key.append(vals.get(18).toString());
+ key.append("");
+ key.append(vals.get(19).toString());
+ key.append("");
+ key.append(vals.get(20).toString());
+ key.append("");
+ key.append(vals.get(21).toString());
+ key.append("");
+ key.append(vals.get(22).toString());
+ key.append("");
+ key.append(vals.get(23).toString());
+ key.append("");
+ key.append(vals.get(24).toString());
+
+ StringBuffer sb = new StringBuffer();
+ sb.append(vals.get(2).toString());
+ sb.append("");
+ sb.append(vals.get(11).toString());
+ sb.append("");
+ sb.append(vals.get(20).toString());
+ sb.append("");
+ sb.append(vals.get(6).toString());
+ sb.append("");
+ sb.append(vals.get(15).toString());
+ sb.append("");
+ sb.append(vals.get(24).toString());
+ oc.collect(new Text(key.toString()), new Text(sb.toString()));
+ }
+ }
+
+ public static class Combiner extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ int tsSum = 0, tsSum1 = 0, tsSum2 = 0, erCnt = 0;
+ double erSum = 0.0, erSum1 = 0.0, erSum2=0.0;
+ while (iter.hasNext()) {
+ List<Text> vals = Library.splitLine(iter.next(), '');
+ try {
+ tsSum += Integer.valueOf(vals.get(0).toString());
+ } catch (NumberFormatException nfe) {
+ }
+ try {
+ tsSum1 += Integer.valueOf(vals.get(1).toString());
+ } catch (NumberFormatException nfe) {
+ }
+ try {
+ tsSum2 += Integer.valueOf(vals.get(2).toString());
+ } catch (NumberFormatException nfe) {
+ }
+ try {
+ erSum += Double.valueOf(vals.get(3).toString());
+ } catch (NumberFormatException nfe) {
+ }
+ try {
+ erSum1 += Double.valueOf(vals.get(4).toString());
+ } catch (NumberFormatException nfe) {
+ }
+ try {
+ erSum2 += Double.valueOf(vals.get(5).toString());
+ } catch (NumberFormatException nfe) {
+ }
+ erCnt++;
+ }
+ StringBuffer sb = new StringBuffer();
+ sb.append((new Integer(tsSum)).toString());
+ sb.append("");
+ sb.append((new Integer(tsSum1)).toString());
+ sb.append("");
+ sb.append((new Integer(tsSum2)).toString());
+ sb.append("");
+ sb.append((new Double(erSum)).toString());
+ sb.append("");
+ sb.append((new Double(erSum1)).toString());
+ sb.append("");
+ sb.append((new Double(erSum2)).toString());
+ sb.append("");
+ sb.append((new Integer(erCnt)).toString());
+ oc.collect(key, new Text(sb.toString()));
+ reporter.setStatus("OK");
+ }
+ }
+ public static class Group extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ int tsSum = 0, tsSum1 = 0, tsSum2 = 0, erCnt = 0;
+ double erSum = 0.0, erSum1 = 0.0, erSum2 = 0.0;
+ while (iter.hasNext()) {
+ List<Text> vals = Library.splitLine(iter.next(), '');
+ try {
+ tsSum += Integer.valueOf(vals.get(0).toString());
+ tsSum1 += Integer.valueOf(vals.get(1).toString());
+ tsSum2 += Integer.valueOf(vals.get(2).toString());
+ erSum += Double.valueOf(vals.get(3).toString());
+ erSum1 += Double.valueOf(vals.get(4).toString());
+ erSum2 += Double.valueOf(vals.get(5).toString());
+ erCnt++;
+ } catch (NumberFormatException nfe) {
+ }
+ }
+ double erAvg = erSum / erCnt, erAvg1 = erSum1 / erCnt, erAvg2 = erSum2 / erCnt;
+ StringBuffer sb = new StringBuffer();
+ sb.append((new Double(tsSum)).toString());
+ sb.append("\t");
+ sb.append((new Double(tsSum1)).toString());
+ sb.append("\t");
+ sb.append((new Double(tsSum2)).toString());
+ sb.append("\t");
+ sb.append((new Double(erAvg)).toString());
+ sb.append("\t");
+ sb.append((new Double(erAvg1)).toString());
+ sb.append("\t");
+ sb.append((new Double(erAvg2)).toString());
+ oc.collect(null, new Text(sb.toString()));
+ reporter.setStatus("OK");
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ JobConf lp = new JobConf(L17.class);
+ lp.setJobName("L17 Wide group by");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ lp.setCombinerClass(Combiner.class);
+ lp.setReducerClass(Group.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/widegroupbydata"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/L17out"));
+ lp.setNumReduceTasks(Integer.parseInt(parallel));
+ Job group = new Job(lp);
+
+ JobControl jc = new JobControl("L17 group by");
+ jc.addJob(group);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L2.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L2.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L2.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L2.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,176 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.BufferedReader;
+import java.io.FileInputStream;
+import java.io.InputStreamReader;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L2 {
+
+ public static class Join extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ Set<String> hash;
+
+ public void configure(JobConf conf) {
+ try {
+ Path[] paths = DistributedCache.getLocalCacheFiles(conf);
+ if (paths == null || paths.length < 1) {
+ throw new RuntimeException("DistributedCache no work.");
+ }
+
+ // Open the small table
+ BufferedReader reader =
+ new BufferedReader(new InputStreamReader(new
+ FileInputStream(paths[0].toString())));
+ String line;
+ hash = new HashSet<String>(500);
+ while ((line = reader.readLine()) != null) {
+ if (line.length() < 1) continue;
+ String[] fields = line.split("");
+ hash.add(fields[0]);
+ }
+ } catch (IOException ioe) {
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ List<Text> fields = Library.splitLine(val, '');
+ String name = fields.get(0).toString();
+
+ String v;
+ if (hash.contains(name)) {
+ StringBuffer sb = new StringBuffer();
+ sb.append(name);
+ sb.append("");
+ sb.append(fields.get(6).toString());
+ oc.collect(null, new Text(sb.toString()));
+ }
+
+ }
+ }
+
+ public static void main(String[] args) throws IOException, URISyntaxException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ JobConf lp = new JobConf(L2.class);
+ lp.setJobName("L2 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(Join.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ DistributedCache.addCacheFile(
+ new URI(inputDir + "/power_users"), lp);
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/L2out"));
+ lp.setNumReduceTasks(0);
+ Job loadPages = new Job(lp);
+
+ JobControl jc = new JobControl("L2 join");
+ jc.addJob(loadPages);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L3.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L3.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L3.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L3.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,225 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L3 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ List<Text> fields = Library.splitLine(val, '');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ Text outVal = new Text("1" + fields.get(6).toString());
+ oc.collect(fields.get(0), outVal);
+ }
+ }
+
+ public static class ReadUsers extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ List<Text> fields = Library.splitLine(val, '');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ Text outVal = new Text("2");
+ oc.collect(fields.get(0), outVal);
+
+ }
+ }
+
+ public static class Join extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ // For each value, figure out which file it's from and store it
+ // accordingly.
+ List<String> first = new ArrayList<String>();
+ List<String> second = new ArrayList<String>();
+
+ while (iter.hasNext()) {
+ Text t = iter.next();
+ String value = t.toString();
+ if (value.charAt(0) == '1') first.add(value.substring(1));
+ else second.add(value.substring(1));
+ reporter.setStatus("OK");
+ }
+
+ reporter.setStatus("OK");
+
+ if (first.size() == 0 || second.size() == 0) return;
+
+ // Do the cross product, and calculate the sum
+ Double sum = 0.0;
+ for (String s1 : first) {
+ for (String s2 : second) {
+ try {
+ sum += Double.valueOf(s1);
+ } catch (NumberFormatException nfe) {
+ }
+ }
+ }
+ oc.collect(null, new Text(key.toString() + "" + sum.toString()));
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ String user = System.getProperty("user.name");
+ JobConf lp = new JobConf(L3.class);
+ lp.setJobName("L3 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/indexed_pages_3"));
+ lp.setNumReduceTasks(0);
+ Job loadPages = new Job(lp);
+
+ JobConf lu = new JobConf(L3.class);
+ lu.setJobName("L3 Load Users");
+ lu.setInputFormat(TextInputFormat.class);
+ lu.setOutputKeyClass(Text.class);
+ lu.setOutputValueClass(Text.class);
+ lu.setMapperClass(ReadUsers.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lu.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lu, new Path(inputDir + "/users"));
+ FileOutputFormat.setOutputPath(lu, new Path(outputDir + "/indexed_users_3"));
+ lu.setNumReduceTasks(0);
+ Job loadUsers = new Job(lu);
+
+ JobConf join = new JobConf(L3.class);
+ join.setJobName("L3 Join Users and Pages");
+ join.setInputFormat(KeyValueTextInputFormat.class);
+ join.setOutputKeyClass(Text.class);
+ join.setOutputValueClass(Text.class);
+ join.setMapperClass(IdentityMapper.class);
+ join.setReducerClass(Join.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ join.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_pages_3"));
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_users_3"));
+ FileOutputFormat.setOutputPath(join, new Path(outputDir + "/L3out"));
+ join.setNumReduceTasks(Integer.parseInt(parallel));
+ Job joinJob = new Job(join);
+ joinJob.addDependingJob(loadPages);
+ joinJob.addDependingJob(loadUsers);
+
+ JobControl jc = new JobControl("L3 join");
+ jc.addJob(loadPages);
+ jc.addJob(loadUsers);
+ jc.addJob(joinJob);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L4.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L4.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L4.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L4.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,168 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L4 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ // Split the line
+ List<Text> fields = Library.splitLine(val, '');
+ if (fields.size() != 9) return;
+
+ oc.collect(fields.get(0), fields.get(1));
+ }
+ }
+
+ public static class Combiner extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ HashSet<Text> hash = new HashSet<Text>();
+ while (iter.hasNext()) {
+ hash.add(iter.next());
+ }
+ for (Text t : hash) oc.collect(key, t);
+ reporter.setStatus("OK");
+ }
+ }
+ public static class Group extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ HashSet<Text> hash = new HashSet<Text>();
+ while (iter.hasNext()) {
+ hash.add(iter.next());
+ }
+ oc.collect(key, new Text(Integer.toString(hash.size())));
+ reporter.setStatus("OK");
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ JobConf lp = new JobConf(L4.class);
+ lp.setJobName("L4 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ lp.setCombinerClass(Combiner.class);
+ lp.setReducerClass(Group.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/L4out"));
+ lp.setNumReduceTasks(Integer.parseInt(parallel));
+ Job group = new Job(lp);
+
+ JobControl jc = new JobControl("L4 join");
+ jc.addJob(group);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}
Added: pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L5.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L5.java?rev=1468268&view=auto
==============================================================================
--- pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L5.java (added)
+++ pig/trunk/test/perf/pigmix/src/java/org/apache/pig/test/pigmix/mapreduce/L5.java Mon Apr 15 23:13:09 2013
@@ -0,0 +1,208 @@
+package org.apache.pig.test.pigmix.mapreduce;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.KeyValueTextInputFormat;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.jobcontrol.Job;
+import org.apache.hadoop.mapred.jobcontrol.JobControl;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+
+import org.apache.pig.test.pigmix.mapreduce.Library;
+
+public class L5 {
+
+ public static class ReadPageViews extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ // Pull the key out
+ List<Text> line = Library.splitLine(val, '');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ oc.collect(line.get(0), new Text("1"));
+ }
+ }
+
+ public static class ReadUsers extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+
+ public void map(
+ LongWritable k,
+ Text val,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+
+ // Pull the key out
+ List<Text> line = Library.splitLine(val, '');
+ // Prepend an index to the value so we know which file
+ // it came from.
+ oc.collect(line.get(0), new Text("2"));
+ }
+ }
+
+ public static class Join extends MapReduceBase
+ implements Reducer<Text, Text, Text, Text> {
+
+ public void reduce(
+ Text key,
+ Iterator<Text> iter,
+ OutputCollector<Text, Text> oc,
+ Reporter reporter) throws IOException {
+ // For each value, figure out which file it's from and store it
+ // accordingly.
+ int cnt = 0;
+
+ while (iter.hasNext()) {
+ if (iter.next().toString().charAt(0) == '2') cnt++;
+ reporter.setStatus("OK");
+ }
+
+ oc.collect(null, key);
+ reporter.setStatus("OK");
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+
+ if (args.length!=3) {
+ System.out.println("Parameters: inputDir outputDir parallel");
+ System.exit(1);
+ }
+ String inputDir = args[0];
+ String outputDir = args[1];
+ String parallel = args[2];
+ String user = System.getProperty("user.name");
+ JobConf lp = new JobConf(L5.class);
+ lp.setJobName("L5 Load Page Views");
+ lp.setInputFormat(TextInputFormat.class);
+ lp.setOutputKeyClass(Text.class);
+ lp.setOutputValueClass(Text.class);
+ lp.setMapperClass(ReadPageViews.class);
+ Properties props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lp.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lp, new Path(inputDir + "/page_views"));
+ FileOutputFormat.setOutputPath(lp, new Path(outputDir + "/indexed_pages_5"));
+ lp.setNumReduceTasks(0);
+ Job loadPages = new Job(lp);
+
+ JobConf lu = new JobConf(L5.class);
+ lu.setJobName("L5 Load Users");
+ lu.setInputFormat(TextInputFormat.class);
+ lu.setOutputKeyClass(Text.class);
+ lu.setOutputValueClass(Text.class);
+ lu.setMapperClass(ReadUsers.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ lu.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(lu, new Path(inputDir + "/users"));
+ FileOutputFormat.setOutputPath(lu, new Path(outputDir + "/indexed_users_5"));
+ lu.setNumReduceTasks(0);
+ Job loadUsers = new Job(lu);
+
+ JobConf join = new JobConf(L5.class);
+ join.setJobName("L5 Join Users and Pages");
+ join.setInputFormat(KeyValueTextInputFormat.class);
+ join.setOutputKeyClass(Text.class);
+ join.setOutputValueClass(Text.class);
+ join.setMapperClass(IdentityMapper.class);
+ join.setReducerClass(Join.class);
+ props = System.getProperties();
+ for (Map.Entry<Object,Object> entry : props.entrySet()) {
+ join.set((String)entry.getKey(), (String)entry.getValue());
+ }
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_pages_5"));
+ FileInputFormat.addInputPath(join, new Path(outputDir + "/indexed_users_5"));
+ FileOutputFormat.setOutputPath(join, new Path(outputDir + "/L5out"));
+ join.setNumReduceTasks(Integer.parseInt(parallel));
+ Job joinJob = new Job(join);
+ joinJob.addDependingJob(loadPages);
+ joinJob.addDependingJob(loadUsers);
+
+ JobControl jc = new JobControl("L5 join");
+ jc.addJob(loadPages);
+ jc.addJob(loadUsers);
+ jc.addJob(joinJob);
+
+ new Thread(jc).start();
+
+ int i = 0;
+ while(!jc.allFinished()){
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ break;
+ }
+
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {}
+
+ if (i % 10000 == 0) {
+ System.out.println("Running jobs");
+ ArrayList<Job> running = jc.getRunningJobs();
+ if (running != null && running.size() > 0) {
+ for (Job r : running) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Ready jobs");
+ ArrayList<Job> ready = jc.getReadyJobs();
+ if (ready != null && ready.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Waiting jobs");
+ ArrayList<Job> waiting = jc.getWaitingJobs();
+ if (waiting != null && waiting.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ System.out.println("Successful jobs");
+ ArrayList<Job> success = jc.getSuccessfulJobs();
+ if (success != null && success.size() > 0) {
+ for (Job r : ready) {
+ System.out.println(r.getJobName());
+ }
+ }
+ }
+ i++;
+ }
+ ArrayList<Job> failures = jc.getFailedJobs();
+ if (failures != null && failures.size() > 0) {
+ for (Job failure : failures) {
+ System.err.println(failure.getMessage());
+ }
+ }
+ jc.stop();
+ }
+
+}