You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrql.apache.org by fe...@apache.org on 2013/10/29 15:29:01 UTC
git commit: MRQL-25: Changed Translator/Evaluator interface to
improve Spark efficiency
Updated Branches:
refs/heads/master 451c3f081 -> 4b096c01b
MRQL-25: Changed Translator/Evaluator interface to improve Spark efficiency
Project: http://git-wip-us.apache.org/repos/asf/incubator-mrql/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-mrql/commit/4b096c01
Tree: http://git-wip-us.apache.org/repos/asf/incubator-mrql/tree/4b096c01
Diff: http://git-wip-us.apache.org/repos/asf/incubator-mrql/diff/4b096c01
Branch: refs/heads/master
Commit: 4b096c01bf43ff070ff964801904a7f28021549b
Parents: 451c3f0
Author: fegaras <fe...@cse.uta.edu>
Authored: Tue Oct 29 09:28:33 2013 -0500
Committer: fegaras <fe...@cse.uta.edu>
Committed: Tue Oct 29 09:28:33 2013 -0500
----------------------------------------------------------------------
src/main/java/BSP/BSPEvaluator.gen | 8 +-
src/main/java/MapReduce/MapReduceEvaluator.gen | 18 +-
src/main/java/core/DataSet.java | 22 +++
src/main/java/core/DataSource.java | 37 ++++
src/main/java/core/Evaluator.java | 64 ++++++-
src/main/java/core/Interpreter.gen | 47 +++--
src/main/java/core/MapReduceAlgebra.java | 145 ++++++++--------
src/main/java/core/Plan.java | 30 +---
src/main/java/core/Printer.gen | 21 ++-
src/main/java/core/Test.java | 12 +-
src/main/java/core/TopLevel.gen | 31 +---
src/main/java/core/Translator.gen | 2 +-
src/main/java/spark/RDDDataSource.java | 30 +++-
src/main/java/spark/SparkEvaluator.gen | 180 +++++++++++---------
14 files changed, 392 insertions(+), 255 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/BSP/BSPEvaluator.gen
----------------------------------------------------------------------
diff --git a/src/main/java/BSP/BSPEvaluator.gen b/src/main/java/BSP/BSPEvaluator.gen
index 2296c67..e8fb859 100644
--- a/src/main/java/BSP/BSPEvaluator.gen
+++ b/src/main/java/BSP/BSPEvaluator.gen
@@ -18,6 +18,7 @@
package org.apache.mrql;
import org.apache.mrql.gen.*;
+import java.util.List;
import java.io.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.conf.Configuration;
@@ -99,11 +100,6 @@ final public class BSPEvaluator extends Evaluator {
throw new Error("Cannot perform the BSP plan: "+plan);
}
- /** collect all data from a persistent dataset into a Bag */
- public Bag collect ( final DataSet x ) throws Exception {
- throw new Error("method collect should be called from BSP");
- }
-
/** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
return BSPParsedInputFormat.class;
@@ -170,7 +166,7 @@ final public class BSPEvaluator extends Evaluator {
if (Config.trace_execution)
try {
System.out.println(tabs(tab_count)
- +"-> "+Plan.collect(res,false));
+ +"-> "+res.take(Config.max_bag_size_print));
tab_count -= 3;
} catch (Exception ex) {
throw new Error("Cannot collect the operator output: "+e);
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/MapReduce/MapReduceEvaluator.gen
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReduceEvaluator.gen b/src/main/java/MapReduce/MapReduceEvaluator.gen
index 3e53a92..e733715 100644
--- a/src/main/java/MapReduce/MapReduceEvaluator.gen
+++ b/src/main/java/MapReduce/MapReduceEvaluator.gen
@@ -54,11 +54,6 @@ final public class MapReduceEvaluator extends Evaluator {
return new Configuration();
}
- /** collect all data from a persistent dataset into a Bag */
- final public Bag collect ( final DataSet x ) throws Exception {
- return MapReducePlan.collect(x);
- }
-
/** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
return MapReduceParsedInputFormat.class;
@@ -114,8 +109,8 @@ final public class MapReduceEvaluator extends Evaluator {
* @return a DataSet (stored in HDFS)
*/
final public DataSet eval ( final Tree e,
- final Environment env,
- final String counter ) {
+ final Environment env,
+ final String counter ) {
if (Config.trace_execution) {
tab_count += 3;
System.out.println(tabs(tab_count)+print_query(e));
@@ -124,7 +119,7 @@ final public class MapReduceEvaluator extends Evaluator {
if (Config.trace_execution)
try {
System.out.println(tabs(tab_count)
- +"-> "+MapReducePlan.collect(res));
+ +"-> "+res.take(Config.max_bag_size_print));
tab_count -= 3;
} catch (Exception ex) {
throw new Error("Cannot collect the operator output: "+e);
@@ -203,9 +198,10 @@ final public class MapReduceEvaluator extends Evaluator {
case GroupByJoin(`kx,`ky,`gx,`gy,`mp,`c,`r,`x,`y,`o):
DataSet X = eval(x,env,"-");
DataSet Y = eval(y,env,"-");
- int n = (int)Math.ceil(X.size(MapReducePlan.conf)/10000.0/Math.sqrt((float)Config.mapjoin_size));
- int m = (int)Math.ceil(Y.size(MapReducePlan.conf)/10000.0/Math.sqrt((float)Config.mapjoin_size));
- System.err.println("*** Using a groupBy join on a "+n+"*"+m+" grid of partitions");
+ int n = (int)Math.floor(Math.sqrt(4.0*Config.nodes));
+ int m = n;
+ if (Config.trace)
+ System.out.println("*** Using a groupBy join on a "+n+"*"+m+" grid of partitions");
return GroupByJoinPlan.groupByJoin(closure(kx,env),closure(ky,env),
closure(gx,env),closure(gy,env),
closure(mp,env),closure(c,env),closure(r,env),
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/DataSet.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/DataSet.java b/src/main/java/core/DataSet.java
index b41d4e2..efe5646 100644
--- a/src/main/java/core/DataSet.java
+++ b/src/main/java/core/DataSet.java
@@ -17,6 +17,7 @@
*/
package org.apache.mrql;
+import java.util.List;
import java.util.ArrayList;
import org.apache.hadoop.conf.Configuration;
@@ -78,6 +79,27 @@ public class DataSet {
return path;
}
+ /** return the first num values */
+ public List<MRData> take ( int num ) {
+ int count = num;
+ ArrayList<MRData> res = new ArrayList<MRData>();
+ for ( DataSource s: source ) {
+ res.addAll(s.take(count));
+ if (res.size() < count)
+ count = count-res.size();
+ else return res;
+ };
+ return res;
+ }
+
+ /** accumulate all dataset values */
+ public MRData reduce ( MRData zero, Function acc ) {
+ MRData res = zero;
+ for ( DataSource s: source )
+ res = s.reduce(res,acc);
+ return res;
+ }
+
public String toString () {
String p = "<"+counter;
for (DataSource s: source)
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/DataSource.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/DataSource.java b/src/main/java/core/DataSource.java
index ffd5f30..6a0ef4f 100644
--- a/src/main/java/core/DataSource.java
+++ b/src/main/java/core/DataSource.java
@@ -19,8 +19,11 @@ package org.apache.mrql;
import org.apache.mrql.gen.*;
import java.io.IOException;
+import java.util.List;
+import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
+import java.util.Iterator;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.Configuration;
@@ -155,4 +158,38 @@ public class DataSource {
dataSourceDirectory.put(local_path,ds);
return ds;
}
+
+ /** return the first num values */
+ public List<MRData> take ( int num ) {
+ int count = num;
+ try {
+ ArrayList<MRData> res = new ArrayList<MRData>();
+ Iterator<MRData> it = inputFormat.newInstance().materialize(new Path(path)).iterator();
+ for ( int i = num; (num < 0 || i > 0) && it.hasNext(); i-- )
+ if (Config.hadoop_mode && Config.bsp_mode)
+ res.add(((Tuple)it.next()).get(1)); // strip tag in BSP mode
+ else res.add(it.next());
+ return res;
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
+
+ static Tuple tuple_container = new Tuple(new Tuple(),new Tuple());
+
+ /** accumulate all datasource values */
+ public MRData reduce ( MRData zero, final Function acc ) {
+ try {
+ MRData res = zero;
+ for ( MRData x: inputFormat.newInstance().materialize(new Path(path)) ) {
+ if (Config.hadoop_mode && Config.bsp_mode)
+ x = ((Tuple)x).get(1); // strip tag in BSP mode
+ tuple_container.set(0,res).set(1,x);
+ res = acc.eval(tuple_container);
+ };
+ return res;
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/Evaluator.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Evaluator.java b/src/main/java/core/Evaluator.java
index bc7488e..28d356f 100644
--- a/src/main/java/core/Evaluator.java
+++ b/src/main/java/core/Evaluator.java
@@ -21,6 +21,8 @@ import java_cup.runtime.*;
import org.apache.mrql.gen.*;
import java.io.*;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
/** Evaluates physical plans using one of the evaluation engines */
@@ -56,9 +58,6 @@ abstract public class Evaluator extends Interpreter {
throw new Error("You can only run a BSP task in BSP mode");
}
- /** collect all data from a persistent dataset into a Bag */
- abstract public Bag collect ( final DataSet x ) throws Exception;
-
/** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
abstract public Class<? extends MRQLFileInputFormat> parsedInputFormat ();
@@ -91,4 +90,63 @@ abstract public class Evaluator extends Interpreter {
abstract public DataSet eval ( final Tree e,
final Environment env,
final String counter );
+
+ final static MR_long counter_key = new MR_long(0);
+ final static MRContainer counter_container = new MRContainer(counter_key);
+ final static MRContainer value_container = new MRContainer(new MR_int(0));
+
+ /** dump MRQL data into a sequence file */
+ public void dump ( String file, Tree type, MRData data ) throws Exception {
+ Path path = new Path(file);
+ FileSystem fs = path.getFileSystem(Plan.conf);
+ PrintStream ftp = new PrintStream(fs.create(path.suffix(".type")));
+ ftp.print("2@"+type.toString()+"\n");
+ ftp.close();
+ SequenceFile.Writer writer
+ = new SequenceFile.Writer(fs,Plan.conf,path,
+ MRContainer.class,MRContainer.class);
+ if (data instanceof MR_dataset)
+ data = Plan.collect(((MR_dataset)data).dataset());
+ if (data instanceof Bag) {
+ Bag s = (Bag)data;
+ long i = 0;
+ for ( MRData e: s ) {
+ counter_key.set(i++);
+ value_container.set(e);
+ writer.append(counter_container,value_container);
+ }
+ } else {
+ counter_key.set(0);
+ value_container.set(data);
+ writer.append(counter_container,value_container);
+ };
+ writer.close();
+ }
+
+ /** dump MRQL data into a text CVS file */
+ public void dump_text ( String file, Tree type, MRData data ) throws Exception {
+ int ps = Config.max_bag_size_print;
+ Config.max_bag_size_print = -1;
+ final PrintStream out = (Config.hadoop_mode)
+ ? Plan.print_stream(file)
+ : new PrintStream(file);
+ if (data instanceof MR_dataset)
+ data = Plan.collect(((MR_dataset)data).dataset());
+ if (Translator.collection_type(type)) {
+ Tree tp = ((Node)type).children().head();
+ if (tp instanceof Node && ((Node)tp).name().equals("tuple")) {
+ Trees ts = ((Node)tp).children();
+ for ( MRData x: (Bag)data ) {
+ Tuple t = (Tuple)x;
+ out.print(print(t.get((short)0),ts.nth(0)));
+ for ( short i = 1; i < t.size(); i++ )
+ out.print(","+print(t.get(i),ts.nth(i)));
+ out.println();
+ }
+ } else for ( MRData x: (Bag)data )
+ out.println(print(x,tp));
+ } else out.println(print(data,query_type));
+ Config.max_bag_size_print = ps;
+ out.close();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/Interpreter.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/Interpreter.gen b/src/main/java/core/Interpreter.gen
index bc8346c..97c76ce 100644
--- a/src/main/java/core/Interpreter.gen
+++ b/src/main/java/core/Interpreter.gen
@@ -18,6 +18,8 @@
package org.apache.mrql;
import org.apache.mrql.gen.*;
+import java.util.List;
+import java.util.ArrayList;
import java.util.Iterator;
import java.io.PrintStream;
@@ -101,8 +103,8 @@ public class Interpreter extends TypeInference {
/** evaluate an MRQL function in memory */
private final static Function evalf ( final String v,
- final Tree body,
- final Environment env ) {
+ final Tree body,
+ final Environment env ) {
return new Function() {
final public MRData eval ( final MRData x ) {
return evalE(body,new Environment(v,x,env));
@@ -346,21 +348,28 @@ public class Interpreter extends TypeInference {
case index(`x,`n):
MRData xv = evalE(x,env);
MRData nv = evalE(n,env);
- if (xv instanceof MR_dataset)
- xv = Plan.collect(((MR_dataset)xv).dataset());
+ final int k = (int)((MR_int)nv).get();
+ if (k < 0)
+ throw new Error("Negative list index: "+k);
+ if (xv instanceof MR_dataset) {
+ List<MRData> res = ((MR_dataset)xv).dataset().take(k+1);
+ if (k >= res.size())
+ throw new Error("List index out of bounds: "+k);
+ return res.get(k);
+ };
Bag b = (Bag)xv;
- int k = (int)((MR_int)nv).get();
return b.get(k);
case range(`x,`i,`j):
MRData xv = evalE(x,env);
MRData ni = evalE(i,env);
MRData nj = evalE(j,env);
- if (xv instanceof MR_dataset)
- xv = Plan.collect(((MR_dataset)xv).dataset());
- Bag b = (Bag)xv;
int ki = (int)((MR_int)ni).get();
int kj = (int)((MR_int)nj).get();
- Iterator<MRData> it = b.iterator();
+ if (ki < 0 || kj < ki)
+ throw new Error("Wrong list range: ["+ki+","+kj+"]");
+ Iterator<MRData> it = (xv instanceof MR_dataset)
+ ? ((MR_dataset)xv).dataset().take(kj+1).iterator()
+ : ((Bag)xv).iterator();
Bag s = new Bag();
for ( int n = 0; it.hasNext() && n < ki; n++ )
it.next();
@@ -369,9 +378,19 @@ public class Interpreter extends TypeInference {
return s;
case map_index(`x,`key):
MRData xv = evalE(x,env);
- MRData nk = evalE(key,env);
- if (xv instanceof MR_dataset)
- xv = Plan.collect(((MR_dataset)xv).dataset());
+ final MRData nk = evalE(key,env);
+ if (xv instanceof MR_dataset) {
+ xv = ((MR_dataset)xv).dataset().reduce(new Tuple(),new Function() {
+ public MRData eval ( MRData value ) {
+ Tuple p = (Tuple)value;
+ Tuple y = (Tuple)p.second();
+ return (y.first().equals(nk)) ? y.second() : p.first();
+ }
+ });
+ if (xv instanceof Tuple && ((Tuple)xv).size() == 0)
+ throw new Error("Map key not found: "+nk);
+ return xv;
+ };
return ((Bag)xv).map_find(nk);
case aggregate(`acc,`zero,`s):
return MapReduceAlgebra.aggregate(evalF(acc,env),evalE(zero,env),
@@ -380,6 +399,10 @@ public class Interpreter extends TypeInference {
if (Config.hadoop_mode)
return Evaluator.evaluator.aggregate(closure(acc,env),zero,s,env);
else return MapReduceAlgebra.aggregate(evalF(acc,env),evalE(zero,env),evalM(s,env));
+ case mergeGroupByJoin(`kx,`ky,`gx,`gy,`m,`c,`r,`x,`y,`o):
+ return MapReduceAlgebra.mergeGroupByJoin(evalF(kx,env),evalF(ky,env),evalF(gx,env),evalF(gy,env),
+ evalF(m,env),evalF(c,env),evalF(r,env),
+ (Bag)evalE(x,env),(Bag)evalE(y,env));
case BSP(tuple(...ns),`superstep,`state,`o,...as):
if (Config.hadoop_mode)
return Evaluator.evaluator.bsp(e,env);
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/MapReduceAlgebra.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MapReduceAlgebra.java b/src/main/java/core/MapReduceAlgebra.java
index 9b50ada..198a533 100644
--- a/src/main/java/core/MapReduceAlgebra.java
+++ b/src/main/java/core/MapReduceAlgebra.java
@@ -207,14 +207,13 @@ final public class MapReduceAlgebra {
/** A hash-based equi-join
* @param kx left key function from a to k
* @param ky right key function from b to k
- * @param f reducer from ({a},{b}) to {c}
+ * @param f reducer from (a,b) to c
* @param X left input of type {a}
* @param Y right input of type {b}
* @return a value of type {c}
*/
public static Bag hash_join ( final Function kx, final Function ky, final Function f,
final Bag X, final Bag Y ) {
- final Bag empty_bag = new Bag();
Hashtable<MRData,Bag> hashTable = new Hashtable<MRData,Bag>(1000);
for ( MRData x: X ) {
MRData key = kx.eval(x);
@@ -227,10 +226,9 @@ final public class MapReduceAlgebra {
for ( MRData y: Y ) {
MRData key = ky.eval(y);
Bag match = hashTable.get(key);
- if (match == null)
- res.add(f.eval(new Tuple(empty_bag,y)));
- else for ( MRData x: match )
- res.add(f.eval(new Tuple(x,y)));
+ if (match != null)
+ for ( MRData x: match )
+ res.add(f.eval(new Tuple(x,y)));
};
return res;
}
@@ -331,7 +329,7 @@ final public class MapReduceAlgebra {
/** The fragment-replicate join (map-side join)
* @param kx left key function from a to k
* @param ky right key function from b to k
- * @param r reducer from ({a},{b}) to {c}
+ * @param r reducer from (a,{b}) to {c}
* @param X left input of type {a}
* @param Y right input of type {b}
* @return a value of type {c}
@@ -388,16 +386,13 @@ final public class MapReduceAlgebra {
return res;
}
- private static void flush_table ( Hashtable<MRData,MRData> hashTable, Function r, Bag result ) throws IOException {
+ private static void flush_table ( final Map<MRData,MRData> hashTable, final Function r, final Bag result ) {
Bag tbag = new Bag(2);
Tuple pair = new Tuple(2);
- Enumeration<MRData> en = hashTable.keys();
- while (en.hasMoreElements()) {
- MRData key = en.nextElement();
- MRData value = hashTable.get(key);
- pair.set(0,key);
+ for ( Map.Entry<MRData,MRData> entry: hashTable.entrySet() ) {
+ pair.set(0,entry.getKey());
tbag.clear();
- tbag.add_element(value);
+ tbag.add_element(entry.getValue());
pair.set(1,tbag);
for ( MRData e: (Bag)r.eval(pair) )
result.add(e);
@@ -417,69 +412,67 @@ final public class MapReduceAlgebra {
* @param Y right input of type {b}
* @return a value of type {e}
*/
- public static Bag mergeGroupByJoin ( final Function kx, final Function ky,
- final Function gx, final Function gy,
- final Function m, final Function c, final Function r,
- final Bag X, final Bag Y ) {
- try {
- Bag tbag = new Bag(2);
- Tuple pair = new Tuple(2);
- Hashtable<MRData,MRData> hashTable = new Hashtable<MRData,MRData>(1000);
- Bag xs = groupBy(map(new Function() {
- public MRData eval ( final MRData e ) {
- Tuple t = (Tuple)e;
- return new Tuple(new Tuple(t.first(),kx.eval(t.second())),t.second());
- } }, X));
- Bag ys = groupBy(map(new Function() {
- public MRData eval ( final MRData e ) {
- Tuple t = (Tuple)e;
- return new Tuple(new Tuple(t.first(),ky.eval(t.second())),t.second());
- } }, Y));
- Bag res = new Bag();
- Iterator<MRData> xi = xs.iterator();
- Iterator<MRData> yi = ys.iterator();
- if ( !xi.hasNext() || !yi.hasNext() )
- return res;
- Tuple x = (Tuple)xi.next();
- Tuple y = (Tuple)yi.next();
- MRData partition = null;
- while ( xi.hasNext() && yi.hasNext() ) {
- int cmp = x.first().compareTo(y.first());
- if (cmp < 0) { x = (Tuple)xi.next(); continue; };
- if (cmp > 0) { y = (Tuple)yi.next(); continue; };
- if (partition == null)
- partition = ((Tuple)x.first()).first();
- else if (!partition.equals(((Tuple)x.first()).first())) {
- partition = ((Tuple)x.first()).first();
- flush_table(hashTable,r,res);
- };
- for ( MRData xx: (Bag)x.second() )
- for ( MRData yy: (Bag)y.second() ) {
- Tuple key = new Tuple(gx.eval(xx),gy.eval(yy));
- Tuple value = new Tuple(xx,yy);
- MRData old = hashTable.get(key);
- pair.set(0,key);
- for ( MRData e: (Bag)m.eval(value) )
- if (old == null)
- hashTable.put(key,e);
- else {
- tbag.clear();
- tbag.add_element(e).add_element(old);
- pair.set(1,tbag);
- for ( MRData z: (Bag)c.eval(pair) )
- hashTable.put(key,z); // normally, done once
- }
- };
- if (xi.hasNext())
- x = (Tuple)xi.next();
- if (yi.hasNext())
- y = (Tuple)yi.next();
- };
- flush_table(hashTable,r,res);
+ final public static Bag mergeGroupByJoin ( final Function kx, final Function ky,
+ final Function gx, final Function gy,
+ final Function m, final Function c, final Function r,
+ Bag X, Bag Y ) {
+ Bag tbag = new Bag(2);
+ Tuple pair = new Tuple(2);
+ Tuple vpair = new Tuple(2);
+ final Map<MRData,MRData> hashTable = new HashMap<MRData,MRData>(1000);
+ Bag xs = groupBy(map(new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(t.first(),kx.eval(t.second())),t.second());
+ } }, X));
+ Bag ys = groupBy(map(new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(t.first(),ky.eval(t.second())),t.second());
+ } }, Y));
+ X = null; Y = null;
+ Bag res = new Bag();
+ final Iterator<MRData> xi = xs.iterator();
+ final Iterator<MRData> yi = ys.iterator();
+ if ( !xi.hasNext() || !yi.hasNext() )
return res;
- } catch (IOException ex) {
- throw new Error(ex);
- }
+ Tuple x = (Tuple)xi.next();
+ Tuple y = (Tuple)yi.next();
+ MRData partition = null;
+ while ( xi.hasNext() && yi.hasNext() ) {
+ int cmp = x.first().compareTo(y.first());
+ if (cmp < 0) { x = (Tuple)xi.next(); continue; };
+ if (cmp > 0) { y = (Tuple)yi.next(); continue; };
+ if (partition == null)
+ partition = ((Tuple)x.first()).first();
+ else if (!partition.equals(((Tuple)x.first()).first())) {
+ partition = ((Tuple)x.first()).first();
+ flush_table(hashTable,r,res);
+ };
+ for ( MRData xx: (Bag)x.second() )
+ for ( MRData yy: (Bag)y.second() ) {
+ Tuple key = new Tuple(gx.eval(xx),gy.eval(yy));
+ vpair.set(0,xx).set(1,yy);
+ MRData old = hashTable.get(key);
+ pair.set(0,key);
+ for ( MRData e: (Bag)m.eval(vpair) )
+ if (old == null)
+ hashTable.put(key,e);
+ else {
+ tbag.clear();
+ tbag.add_element(e).add_element(old);
+ pair.set(1,tbag);
+ for ( MRData z: (Bag)c.eval(pair) )
+ hashTable.put(key,z); // normally, done once
+ }
+ };
+ if (xi.hasNext())
+ x = (Tuple)xi.next();
+ if (yi.hasNext())
+ y = (Tuple)yi.next();
+ };
+ flush_table(hashTable,r,res);
+ return res;
}
/** repeat the loop until all termination conditions are true or until we reach the max num of steps
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/Plan.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Plan.java b/src/main/java/core/Plan.java
index 6ecdafa..c4951bf 100644
--- a/src/main/java/core/Plan.java
+++ b/src/main/java/core/Plan.java
@@ -27,7 +27,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.SequenceFile.Sorter;
-/** A physical plan (a superclass for both MapReduce and BSP plans) */
+/** A physical plan (a superclass for both MapReduce, BSP, and Spark plans) */
public class Plan {
public static Configuration conf;
static ArrayList<String> temporary_paths = new ArrayList<String>();
@@ -359,32 +359,4 @@ public class Plan {
FileSystem fs = path.getFileSystem(conf);
return new PrintStream(fs.create(path));
}
-
- /** dump both the MRQL data and their MRQL type into files */
- public final static void dump ( String file, Tree type, MRData data ) throws Exception {
- Path path = new Path(file);
- FileSystem fs = path.getFileSystem(conf);
- PrintStream ftp = new PrintStream(fs.create(path.suffix(".type")));
- ftp.print("2@"+type.toString()+"\n");
- ftp.close();
- SequenceFile.Writer writer
- = new SequenceFile.Writer(fs,conf,path,
- MRContainer.class,MRContainer.class);
- if (data instanceof MR_dataset)
- data = Plan.collect(((MR_dataset)data).dataset());
- if (data instanceof Bag) {
- Bag s = (Bag)data;
- long i = 0;
- for ( MRData e: s ) {
- counter_key.set(i++);
- value_container.set(e);
- writer.append(counter_container,value_container);
- }
- } else {
- counter_key.set(0);
- value_container.set(data);
- writer.append(counter_container,value_container);
- };
- writer.close();
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/Printer.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/Printer.gen b/src/main/java/core/Printer.gen
index 0ed282c..d0d80da 100644
--- a/src/main/java/core/Printer.gen
+++ b/src/main/java/core/Printer.gen
@@ -19,6 +19,7 @@ package org.apache.mrql;
import org.apache.mrql.gen.*;
import java.util.Iterator;
+import java.util.List;
/** printers for types, expressions, plans, etc */
@@ -202,12 +203,28 @@ public class Printer {
case Bag(`tp):
if (x instanceof MR_dataset) {
DataSet ds = ((MR_dataset)x).dataset();
- return print(Plan.collect(ds),#<bag(`tp)>);
+ List<MRData> vals = ds.take(Config.max_bag_size_print);
+ if (vals.size() == 0)
+ return "{}";
+ String s = "{ "+print(vals.get(0),tp);
+ for ( int i = 1; i < vals.size(); i++ )
+ s += ", "+print(vals.get(i),tp);
+ if (vals.size() == Config.max_bag_size_print)
+ return s+", ... }";
+ else return s+" }";
} else return print(x,#<bag(`tp)>);
case List(`tp):
if (x instanceof MR_dataset) {
DataSet ds = ((MR_dataset)x).dataset();
- return print(Plan.collect(ds),#<list(`tp)>);
+ List<MRData> vals = ds.take(Config.max_bag_size_print);
+ if (vals.size() == 0)
+ return "[]";
+ String s = "[ "+print(vals.get(0),tp);
+ for ( int i = 1; i < vals.size(); i++ )
+ s += ", "+print(vals.get(i),tp);
+ if (vals.size() == Config.max_bag_size_print)
+ return s+", ... ]";
+ else return s+" ]";
} else return print(x,#<list(`tp)>);
case bag(`tp):
Bag b = (Bag)x;
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/Test.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Test.java b/src/main/java/core/Test.java
index 78cb0d5..eca272c 100644
--- a/src/main/java/core/Test.java
+++ b/src/main/java/core/Test.java
@@ -29,6 +29,7 @@ final public class Test {
static MRQLParser parser = new MRQLParser();
static String result_directory;
static PrintStream test_out;
+ static PrintStream error_stream;
private static int compare ( String file1, String file2 ) throws Exception {
FileInputStream s1 = new FileInputStream(file1);
@@ -65,14 +66,14 @@ final public class Test {
test_out.println("OK matched");
else test_out.println("OK created");
} catch (Error ex) {
- System.err.println(qname+": "+ex);
- ex.printStackTrace(System.err);
+ error_stream.println(qname+": "+ex);
+ ex.printStackTrace(error_stream);
test_out.println("FAILED");
if (!exists)
new File(result_file).delete();
} catch (Exception ex) {
- System.err.println(qname+": "+ex);
- ex.printStackTrace(System.err);
+ error_stream.println(qname+": "+ex);
+ ex.printStackTrace(error_stream);
test_out.println("FAILED");
if (!exists)
new File(result_file).delete();
@@ -117,7 +118,8 @@ final public class Test {
File query_dir = new File(Main.query_file);
result_directory = Config.extra_args.get(0);
(new File(result_directory)).mkdirs();
- System.setErr(new PrintStream(Config.extra_args.get(1)));
+ error_stream = new PrintStream(Config.extra_args.get(1));
+ System.setErr(error_stream);
test_out = System.out;
for ( File f: query_dir.listFiles() )
query(f);
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/TopLevel.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/TopLevel.gen b/src/main/java/core/TopLevel.gen
index 9972b86..0ed3f49 100644
--- a/src/main/java/core/TopLevel.gen
+++ b/src/main/java/core/TopLevel.gen
@@ -18,7 +18,6 @@
package org.apache.mrql;
import org.apache.mrql.gen.*;
-import java.io.PrintStream;
/** Provides the API for compilation/code-generation */
@@ -76,12 +75,16 @@ final public class TopLevel extends Interpreter {
System.err.println("*** MRQL System Error at line "+Main.parser.line_pos()+": "+x);
if (Config.trace)
x.printStackTrace(System.err);
+ if (Config.testing)
+ throw new Error(x);
return null;
} catch (Error x) {
if (x.getMessage() != null)
System.err.println("*** MRQL System Error at line "+Main.parser.line_pos()+": "+x);
if (Config.trace)
x.printStackTrace(System.err);
+ if (Config.testing)
+ throw new Error(x);
return null;
}
}
@@ -172,7 +175,7 @@ final public class TopLevel extends Interpreter {
query_type = make_persistent_type(query_type);
if (res != null)
if (Config.hadoop_mode)
- Plan.dump(file,query_type,res);
+ Evaluator.evaluator.dump(file,query_type,res);
else MapReduceAlgebra.dump(file,query_type,res);
} catch (Exception x) {
throw new Error(x);
@@ -184,29 +187,7 @@ final public class TopLevel extends Interpreter {
MRData res = expression(e,false);
if (res != null)
try {
- PrintStream out = (Config.hadoop_mode)
- ? Plan.print_stream(file)
- : new PrintStream(file);
- if (res instanceof MR_dataset)
- res = Plan.collect(((MR_dataset)res).dataset());
- int ps = Config.max_bag_size_print;
- Config.max_bag_size_print = -1;
- match query_type {
- case `T(tuple(...ts)):
- for ( MRData x: (Bag)res ) {
- Tuple t = (Tuple)x;
- out.print(print(t.get((short)0),ts.nth(0)));
- for ( short i = 1; i < t.size(); i++ )
- out.print(","+print(t.get(i),ts.nth(i)));
- out.println();
- }
- case `T(`tp):
- for ( MRData x: (Bag)res )
- out.println(print(x,tp));
- case _: out.println(print(res,query_type));
- };
- Config.max_bag_size_print = ps;
- out.close();
+ Evaluator.evaluator.dump_text(file,query_type,res);
} catch (Exception x) {
throw new Error(x);
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/core/Translator.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/Translator.gen b/src/main/java/core/Translator.gen
index 19a7a5c..47b34ac 100644
--- a/src/main/java/core/Translator.gen
+++ b/src/main/java/core/Translator.gen
@@ -178,7 +178,7 @@ public class Translator extends Printer {
static Trees plans_with_distributed_lambdas
= #[MapReduce,MapAggregateReduce,MapCombineReduce,FroupByJoin,Aggregate,
MapReduce2,MapCombineReduce2,MapAggregateReduce2,MapJoin,MapAggregateJoin,
- CrossProduct,CrossAggregateProduct,cMap,AggregateMap,BSP,repeat,closure];
+ CrossProduct,CrossAggregateProduct,cMap,AggregateMap,BSP,GroupByJoin,repeat,closure];
static Trees algebraic_operators
= #[mapReduce,mapReduce2,cmap,join,groupBy,orderBy,aggregate,map,filter];
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/spark/RDDDataSource.java
----------------------------------------------------------------------
diff --git a/src/main/java/spark/RDDDataSource.java b/src/main/java/spark/RDDDataSource.java
index 0e59a29..b0345da 100644
--- a/src/main/java/spark/RDDDataSource.java
+++ b/src/main/java/spark/RDDDataSource.java
@@ -17,12 +17,15 @@
*/
package org.apache.mrql;
+import java.util.List;
+import java.io.Serializable;
import org.apache.hadoop.conf.Configuration;
import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function2;
-final public class RDDDataSource extends DataSource {
- JavaRDD<MRData> rdd;
+final public class RDDDataSource extends DataSource implements Serializable {
+ public JavaRDD<MRData> rdd;
RDDDataSource ( JavaRDD<MRData> rdd ) {
super();
@@ -33,4 +36,27 @@ final public class RDDDataSource extends DataSource {
public long size ( Configuration conf ) {
return rdd.count();
}
+
+ /** return the first num values */
+ @Override
+ public List<MRData> take ( int num ) {
+ return (num < 0) ? rdd.collect() : rdd.take(num);
+ }
+
+ /** accumulate all dataset values */
+ @Override
+ public MRData reduce ( final MRData zero, final Function acc ) {
+ return rdd.aggregate(zero,new Function2<MRData,MRData,MRData>() {
+ Tuple t = new Tuple(new Tuple(),new Tuple());
+ public MRData call ( MRData x, MRData y ) {
+ t.set(0,x).set(1,y);
+ return acc.eval(t);
+ }
+ },new Function2<MRData,MRData,MRData>() {
+ Tuple t = new Tuple(new Tuple(),new Tuple());
+ public MRData call ( MRData x, MRData y ) {
+ return (y.equals(zero)) ? x : y;
+ }
+ });
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/4b096c01/src/main/java/spark/SparkEvaluator.gen
----------------------------------------------------------------------
diff --git a/src/main/java/spark/SparkEvaluator.gen b/src/main/java/spark/SparkEvaluator.gen
index 0f0be0f..fbc08ff 100644
--- a/src/main/java/spark/SparkEvaluator.gen
+++ b/src/main/java/spark/SparkEvaluator.gen
@@ -17,10 +17,13 @@
*/
package org.apache.mrql;
-import java_cup.runtime.*;
import org.apache.mrql.gen.*;
import java_cup.runtime.Scanner;
-import java.util.*;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.Iterator;
import java.io.*;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -45,7 +48,7 @@ import org.apache.spark.api.java.function.VoidFunction;
/** Evaluates physical plans in Apache Spark mode */
final public class SparkEvaluator extends Evaluator implements Serializable {
- static JavaSparkContext spark_context;
+ public static JavaSparkContext spark_context;
final static String spark_jar = SparkEvaluator.class.getProtectionDomain().getCodeSource().getLocation().toString();
final static String core_jar = Interpreter.class.getProtectionDomain().getCodeSource().getLocation().toString();
final static String gen_jar = Tree.class.getProtectionDomain().getCodeSource().getLocation().toString();
@@ -85,11 +88,6 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
}
final public void initialize_query () {
- spark_context.clearJars();
- spark_context.addJar(spark_jar);
- spark_context.addJar(core_jar);
- spark_context.addJar(gen_jar);
- spark_context.addJar(scanner_jar);
Plan.distribute_compiled_arguments(Plan.conf);
if (Config.compile_functional_arguments)
spark_context.addJar(Plan.conf.get("mrql.jar.path"));
@@ -99,47 +97,6 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
return new Configuration();
}
- final public MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
- throw new Error("You can only synchronize BSP tasks");
- }
-
- final public Bag distribute ( MR_string peerName, Bag s ) {
- throw new Error("You can only distribute bags among BSP tasks");
- }
-
- final public Bag collect ( final DataSet x ) throws Exception {
- return Plan.collect(x);
- }
-
- final public MRData bsp ( Tree plan, Environment env ) throws Exception {
- throw new Error("You can not run a BSP task in Spark mode");
- }
-
- /** materialize the entire dataset into a Bag
- * @param x the DataSet in HDFS to collect values from
- * @param strip is not used in MapReduce mode
- * @return the Bag that contains the collected values
- */
- final public Bag collect ( final DataSet x, boolean strip ) throws Exception {
- Bag res = new Bag();
- final MRContainer zero = new MRContainer(new MR_byte((byte)0));
- for ( DataSource s: x.source )
- if (s instanceof RDDDataSource) {
- JavaPairRDD<MRContainer,MRContainer> rd
- = ((RDDDataSource)s).rdd.map(new PairFunction<MRData,MRContainer,MRContainer>() {
- public Tuple2<MRContainer,MRContainer> call ( final MRData value ) {
- return new Tuple2<MRContainer,MRContainer>(zero,new MRContainer(value));
- }
- });
- String path = Plan.new_path(Plan.conf);
- rd.saveAsHadoopFile(path,MRContainer.class,MRContainer.class,SequenceFileOutputFormat.class);
- res = res.union(collect(new DataSet(new BinaryDataSource(path,Plan.conf),0,0),strip));
- } else if (s.to_be_merged)
- res = res.union(Plan.merge(s));
- else res = res.union(s.inputFormat.newInstance().materialize(new Path(s.path)));
- return res;
- }
-
/** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
return SparkParsedInputFormat.class;
@@ -171,8 +128,11 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
public static void load_source_dir () throws IOException {
if (Config.local_mode)
return;
- if (Plan.conf == null)
- Plan.conf = Evaluator.evaluator.new_configuration();
+ if (Plan.conf == null) {
+ if (evaluator == null)
+ evaluator = new SparkEvaluator();
+ Plan.conf = evaluator.new_configuration();
+ };
// the name of the file that contains the source directory details is passed
// to workers through the HashMap environment in the JavaSparkContext
Path path = new Path(System.getenv("mrql.data.source.directory"));
@@ -183,7 +143,55 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
ftp.close();
}
- private static Function2<MRData,MRData,MRData> accumulator ( Tree acc_fnc, Environment env ) {
+ /** dump MRQL data into a sequence file */
+ @Override
+ final public void dump ( String file, Tree type, MRData data ) throws Exception {
+ if (data instanceof MR_dataset && ((MR_dataset)data).dataset().source.get(0) instanceof RDDDataSource) {
+ Path path = new Path(file);
+ FileSystem fs = path.getFileSystem(Plan.conf);
+ PrintStream ftp = new PrintStream(fs.create(path.suffix(".type")));
+ ftp.print("2@"+type.toString()+"\n");
+ ftp.close();
+ // in Spark mode, a dataset has always one RDD datasource
+ JavaRDD<MRData> rdd = ((RDDDataSource)((MR_dataset)data).dataset().source.get(0)).rdd.cache();
+ rdd.map(new PairFunction<MRData,MRContainer,MRContainer>() {
+ final MRContainer zero = new MRContainer(new MR_int(0));
+ public Tuple2<MRContainer,MRContainer> call ( MRData value ) {
+ return new Tuple2<MRContainer,MRContainer>(zero,new MRContainer(value));
+ }
+ })
+ .saveAsHadoopFile(file,MRContainer.class,MRContainer.class,SequenceFileOutputFormat.class);
+ } else super.dump(file,type,data);
+ }
+
+ /** dump MRQL data into text a CVS file */
+ @Override
+ final public void dump_text ( String file, Tree type, MRData data ) throws Exception {
+ if (data instanceof MR_dataset && ((MR_dataset)data).dataset().source.get(0) instanceof RDDDataSource) {
+ // in Spark mode, a dataset has always one RDD datasource
+ JavaRDD<MRData> rdd = ((RDDDataSource)((MR_dataset)data).dataset().source.get(0)).rdd.cache();
+ int ps = Config.max_bag_size_print;
+ Config.max_bag_size_print = -1;
+ match type {
+ case `T(tuple(...tps)):
+ final Trees ts = tps;
+ rdd.map(new Function<MRData,String>() {
+ public String call ( MRData value ) {
+ Tuple t = (Tuple)value;
+ String s = Printer.print(t.get((short)0),ts.nth(0));
+ for ( short i = 1; i < t.size(); i++ )
+ s += ","+Printer.print(t.get(i),ts.nth(i));
+ return s;
+ }
+ }).saveAsTextFile(file);
+ case _:
+ rdd.saveAsTextFile(file);
+ };
+ Config.max_bag_size_print = ps;
+ } else super.dump_text(file,type,data);
+ }
+
+ private static Function2<MRData,MRData,MRData> accumulator ( final Tree acc_fnc, final Environment env ) {
final org.apache.mrql.Function f = evalF(acc_fnc,env);
return new Function2<MRData,MRData,MRData>() {
public MRData call ( MRData x, MRData y ) {
@@ -339,7 +347,7 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
final JavaRDD<MRData> res = evalD(e,env);
if (Config.trace_execution)
try {
- System.out.println(tabs(tab_count)+"-> "+res.collect());
+ System.out.println(tabs(tab_count)+"-> "+res.take(Config.max_bag_size_print));
tab_count -= 3;
} catch (Exception ex) {
throw new Error("Cannot collect the operator output: "+e);
@@ -394,7 +402,6 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
private static JavaRDD<MRData> containerData ( JavaPairRDD<MRContainer,MRContainer> rd ) {
final Environment env = Interpreter.global_env;
return rd.map(new Function<Tuple2<MRContainer,MRContainer>,MRData>() {
- // need to pass the global bindings (the in-memory repeat vars) to workers
boolean first = true;
public MRData call ( Tuple2<MRContainer,MRContainer> value ) {
if (first) {
@@ -522,8 +529,10 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
}
});
case GroupByJoin(`kx,`ky,`gx,`gy,`mp,`c,`r,`x,`y,`o):
- final int n = 10;
- final int m = 10;
+ final int n = (int)Math.floor(Math.sqrt(4.0*Config.nodes));
+ final int m = n;
+ if (Config.trace)
+ System.out.println("*** Using a groupBy join on a "+n+"*"+m+" grid of partitions");
final org.apache.mrql.Function fkx = evalF(kx,env);
final org.apache.mrql.Function fky = evalF(ky,env);
final org.apache.mrql.Function fgx = evalF(gx,env);
@@ -531,7 +540,9 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
final org.apache.mrql.Function fm = evalF(mp,env);
final org.apache.mrql.Function fc = evalF(c,env);
final org.apache.mrql.Function fr = evalF(r,env);
- JavaPairRDD<MRData,MRData> xs
+ final MRData one = new MR_byte(1);
+ final MRData two = new MR_byte(2);
+ final JavaPairRDD<MRData,MRData> xs
= eval(x,env).flatMap(new PairFlatMapFunction<MRData,MRData,MRData>() {
public Iterable<Tuple2<MRData,MRData>> call ( final MRData value ) {
return new Iterable<Tuple2<MRData,MRData>>() {
@@ -541,7 +552,7 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
public Tuple2<MRData,MRData> next () {
MRData key = new MR_int((fgx.eval(value).hashCode() % m)+m*i);
i++;
- return new Tuple2<MRData,MRData>(key,new Tuple(fkx.eval(value),value));
+ return new Tuple2<MRData,MRData>(key,new Tuple(one,value));
}
public boolean hasNext () {
return i < n;
@@ -552,20 +563,20 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
};
}
});
- JavaPairRDD<MRData,MRData> ys
+ final JavaPairRDD<MRData,MRData> ys
= eval(y,env).flatMap(new PairFlatMapFunction<MRData,MRData,MRData>() {
public Iterable<Tuple2<MRData,MRData>> call ( final MRData value ) {
return new Iterable<Tuple2<MRData,MRData>>() {
public Iterator<Tuple2<MRData,MRData>> iterator() {
return new Iterator<Tuple2<MRData,MRData>>() {
- int i = 0;
+ int j = 0;
public Tuple2<MRData,MRData> next () {
- MRData key = new MR_int((fgy.eval(value).hashCode() % n)*m+i);
- i++;
- return new Tuple2<MRData,MRData>(key,new Tuple(fky.eval(value),value));
+ MRData key = new MR_int((fgy.eval(value).hashCode() % n)*m+j);
+ j++;
+ return new Tuple2<MRData,MRData>(key,new Tuple(two,value));
}
public boolean hasNext () {
- return i < m;
+ return j < m;
}
public void remove () {}
};
@@ -573,25 +584,28 @@ final public class SparkEvaluator extends Evaluator implements Serializable {
};
}
});
- return xs.cogroup(ys)
- .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<List<MRData>,List<MRData>>>,MRData>() {
- public Iterable<MRData> call ( Tuple2<MRData,Tuple2<List<MRData>,List<MRData>>> value ) {
- final Iterator<MRData> i = (MapReduceAlgebra.mergeGroupByJoin(fkx,fky,fgx,fgy,fm,fc,fr,
- bag(value._2._1),bag(value._2._2))).iterator();
- return new Iterable<MRData>() {
- public Iterator<MRData> iterator() {
- return new Iterator<MRData>() {
- public MRData next () {
- return i.next();
- }
- public boolean hasNext () {
- return i.hasNext();
- }
- public void remove () {}
- };
- }
- };
- }
+ return xs.union(ys).groupByKey(Config.nodes)
+ .mapPartitions(new FlatMapFunction<Iterator<Tuple2<MRData,List<MRData>>>,MRData>() {
+ public Iterable<MRData> call ( final Iterator<Tuple2<MRData,List<MRData>>> value ) {
+ Bag xb = new Bag();
+ Bag yb = new Bag();
+ while (value.hasNext()) {
+ Tuple2<MRData,List<MRData>> t = value.next();
+ for ( MRData e: t._2 ) {
+ Tuple p = (Tuple)e;
+ if (((MR_byte)p.first()).get() == 1)
+ xb.add(new Tuple(t._1,p.second()));
+ else yb.add(new Tuple(t._1,p.second()));
+ }
+ };
+ final Bag b = MapReduceAlgebra.mergeGroupByJoin(fkx,fky,fgx,fgy,fm,fc,fr,xb,yb);
+ xb = null; yb = null;
+ return new Iterable<MRData>() {
+ public Iterator<MRData> iterator() {
+ return b.iterator();
+ }
+ };
+ }
});
case MapJoin(`mx,`my,`r,`x,`y):
final org.apache.mrql.Function fx = evalF(mx,env);