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 2014/03/13 15:24:40 UTC
[15/26] MRQL-32: Refactoring directory structure for Eclipse
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/JoinOperation.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/JoinOperation.java b/mapreduce/src/main/java/org/apache/mrql/JoinOperation.java
new file mode 100644
index 0000000..0f962df
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/JoinOperation.java
@@ -0,0 +1,405 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.io.*;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Enumeration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+
+
+/** The MapReduce2 physical operator (a reduce-side join) */
+final public class JoinOperation extends MapReducePlan {
+
+ /** Container for join input values. For reduce-side join, we concatenate
+ * 1 or 2 to the join key before we mix the right with the left tuples */
+ public final static class JoinKey implements Writable {
+ public byte tag; // 1 or 2
+ public MRData key;
+
+ JoinKey () {}
+ JoinKey ( byte t, MRData k ) { tag = t; key = k; }
+
+ public void write ( DataOutput out ) throws IOException {
+ out.writeByte(tag);
+ key.write(out);
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ tag = in.readByte();
+ key = MRContainer.read(in);
+ }
+
+ public String toString () {
+ return "<"+tag+":"+key+">";
+ }
+ }
+
+ public final static class MRContainerJoinPartitioner extends Partitioner<JoinKey,MRContainer> {
+ final public int getPartition ( JoinKey key, MRContainer value, int numPartitions ) {
+ return Math.abs(key.key.hashCode()) % numPartitions;
+ }
+ }
+
+ /** The sorting of the joined values uses ths join key for major order and tag for minor order */
+ public final static class MRContainerSortComparator implements RawComparator<JoinKey> {
+ int[] container_size;
+
+ public MRContainerSortComparator () {
+ container_size = new int[1];
+ }
+
+ final public int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl ) {
+ int cmp = MRContainer.compare(x,xs+1,xl-1,y,ys+1,yl-1,container_size);
+ return (cmp == 0) ? x[xs]-y[ys] : cmp;
+ }
+
+ final public int compare ( JoinKey x, JoinKey y ) {
+ int c = x.key.compareTo(y.key);
+ return (c == 0) ? x.tag-y.tag : c;
+ }
+ }
+
+ /** The grouping of the joined values is based on join key only */
+ public final static class MRContainerGroupingComparator implements RawComparator<JoinKey> {
+ int[] container_size;
+
+ public MRContainerGroupingComparator() {
+ container_size = new int[1];
+ }
+
+ final public int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl ) {
+ return MRContainer.compare(x,xs+1,xl-1,y,ys+1,yl-1,container_size);
+ }
+
+ final public int compare ( JoinKey x, JoinKey y ) {
+ return x.key.compareTo(y.key);
+ }
+ }
+
+ /** The left mapper for MapReduce2 */
+ private final static class MapperLeft extends Mapper<MRContainer,MRContainer,JoinKey,MRContainer> {
+ private static Function mx; // the left mapper function
+ private static JoinKey join_key = new JoinKey((byte)2,new MR_int(0));
+ private static Tuple tvalue = (new Tuple(2)).set(0,new MR_byte(2));
+ private static MRContainer cvalue = new MRContainer(tvalue);
+
+ @Override
+ public void map ( MRContainer key, MRContainer value, Context context )
+ throws IOException, InterruptedException {
+ for ( MRData e: (Bag)mx.eval(value.data()) ) {
+ Tuple p = (Tuple)e;
+ join_key.key = p.first();
+ tvalue.set(1,p.second());
+ cvalue.set(tvalue);
+ context.write(join_key,cvalue);
+ }
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ Tree code = Tree.parse(conf.get("mrql.mapper.left"));
+ mx = functional_argument(conf,code);
+ } catch (Exception e) {
+ throw new Error("Cannot retrieve the left mapper plan");
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ super.cleanup(context);
+ }
+ }
+
+ /** The right mapper for MapReduce2 */
+ private final static class MapperRight extends Mapper<MRContainer,MRContainer,JoinKey,MRContainer> {
+ private static Function my; // the right mapper function
+ private static JoinKey join_key = new JoinKey((byte)1,new MR_int(0));
+ private static Tuple tvalue = (new Tuple(2)).set(0,new MR_byte(1));
+ private static MRContainer cvalue = new MRContainer(tvalue);
+
+ @Override
+ public void map ( MRContainer key, MRContainer value, Context context )
+ throws IOException, InterruptedException {
+ for ( MRData e: (Bag)my.eval(value.data()) ) {
+ Tuple p = (Tuple)e;
+ join_key.key = p.first();
+ tvalue.set(1,p.second());
+ cvalue.set(tvalue);
+ context.write(join_key,cvalue);
+ }
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ Tree code = Tree.parse(conf.get("mrql.mapper.right"));
+ my = functional_argument(conf,code);
+ } catch (Exception e) {
+ throw new Error("Cannot retrieve the right mapper plan");
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ super.cleanup(context);
+ }
+ }
+
+ /** The reducer for MapReduce2 */
+ private static class JoinReducer extends Reducer<JoinKey,MRContainer,MRContainer,MRContainer> {
+ private static String counter; // a Hadoop user-defined counter used in the repeat operation
+ private static Function combine_fnc; // the combiner function
+ private static Function reduce_fnc; // the reduce function
+ private static Bag left = new Bag(); // a cached bag of input fragments from left input
+ private static Bag right = new Bag(); // a cached bag of input fragments from right input
+ private static Function acc_fnc; // aggregator
+ private static MRData result; // aggregation result
+ private static Hashtable<MRData,MRData> hashTable; // in-mapper combiner
+ private static int index;
+ private static Tuple tkey = new Tuple(2);
+ private static Bag tbag = new Bag(2);
+ private static boolean streamed = false;
+ private static Tuple pair = new Tuple(2);
+ private static MRContainer ckey = new MRContainer(new MR_int(0));
+ private static MRContainer cvalue = new MRContainer(new MR_int(0));
+ private static MRContainer container = new MRContainer(new MR_int(0));
+
+ private void write ( MRContainer key, MRData value, Context context )
+ throws IOException, InterruptedException {
+ if (result != null) { // aggregation
+ pair.set(0,result);
+ pair.set(1,value);
+ result = acc_fnc.eval(pair);
+ } else if (hashTable != null) {
+ MRData k = ((Tuple)value).get(0);
+ MRData v = ((Tuple)value).get(1);
+ MRData old = hashTable.get(k);
+ if (old == null) {
+ if (index++ == Config.map_cache_size)
+ flush_table(context);
+ hashTable.put(k,v);
+ } else {
+ tkey.set(0,key.data());
+ tbag.clear();
+ tbag.add_element(v).add_element(old);
+ tkey.set(1,tbag);
+ for ( MRData x: (Bag)combine_fnc.eval(tkey) )
+ hashTable.put(k,x); // normally, done once
+ }
+ } else if (counter.equals("-")) {
+ container.set(value);
+ context.write(key,container);
+ } else { // increment the repetition counter if the repeat condition is true
+ Tuple t = (Tuple)value;
+ if (((MR_bool)t.second()).get())
+ context.getCounter("mrql",counter).increment(1);
+ container.set(t.first());
+ context.write(key,container);
+ }
+ }
+
+ protected static void flush_table ( Context context ) throws IOException, InterruptedException {
+ Enumeration<MRData> en = hashTable.keys();
+ while (en.hasMoreElements()) {
+ MRData key = en.nextElement();
+ ckey.set(key);
+ MRData value = hashTable.get(key);
+ cvalue.set(value);
+ if (value != null)
+ context.write(ckey,new MRContainer(new Tuple(key,value)));
+ };
+ index = 0;
+ hashTable.clear();
+ }
+
+ @Override
+ public void reduce ( JoinKey key, Iterable<MRContainer> values, Context context )
+ throws IOException, InterruptedException {
+ if (!streamed) {
+ left.clear();
+ right.clear();
+ for ( MRContainer val: values ) {
+ Tuple p = (Tuple)val.data();
+ if (((MR_byte)p.first()).get() == 1)
+ right.add(p.second());
+ else left.add(p.second());
+ };
+ } else { // the left input is processed lazily (as a stream-based bag)
+ right.clear();
+ Tuple p = null;
+ final Iterator<MRContainer> i = values.iterator();
+ while (i.hasNext()) {
+ p = (Tuple)i.next().data();
+ if (((MR_byte)p.first()).get() == 2)
+ break;
+ right.add(p.second());
+ p = null;
+ };
+ final Tuple data = p;
+ left = new Bag(new BagIterator () {
+ boolean first_time = data != null;
+ public boolean hasNext () {
+ return first_time || i.hasNext();
+ }
+ public MRData next () {
+ if (!first_time) {
+ Tuple t = (Tuple)i.next().data();
+ assert(((MR_byte)(t.first())).get() == 2);
+ return t.second();
+ };
+ first_time = false;
+ return data.second();
+ }
+ });
+ };
+ pair.set(0,left);
+ pair.set(1,right);
+ for ( MRData e: (Bag)reduce_fnc.eval(pair) ) {
+ ckey.set(key.key);
+ write(ckey,e,context);
+ }
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ Tree code = Tree.parse(conf.get("mrql.reducer"));
+ reduce_fnc = functional_argument(conf,code);
+ streamed = PlanGeneration.streamed_MapReduce2_reducer(code);
+ if (conf.get("mrql.zero") != null) {
+ code = Tree.parse(conf.get("mrql.zero"));
+ result = Interpreter.evalE(code);
+ code = Tree.parse(conf.get("mrql.accumulator"));
+ acc_fnc = functional_argument(conf,code);
+ } else result = null;
+ counter = conf.get("mrql.counter");
+ code = Tree.parse(conf.get("mrql.combiner"));
+ hashTable = null;
+ if (code != null && !code.equals(new VariableLeaf("null"))) {
+ combine_fnc = functional_argument(conf,code);
+ hashTable = new Hashtable<MRData,MRData>(Config.map_cache_size);
+ index = 0;
+ }
+ } catch (Exception e) {
+ throw new Error("Cannot retrieve the reducer plan");
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ if (result != null) // emit the result of aggregation
+ context.write(new MRContainer(new MR_int(0)),new MRContainer(result));
+ if (hashTable != null)
+ flush_table(context);
+ hashTable = null; // garbage-collect it
+ super.cleanup(context);
+ }
+ }
+
+ /** The MapReduce2 physical operator (a reduce-side join)
+ * @param mx left mapper function
+ * @param my right mapper function
+ * @param combine_fnc optional in-mapper combiner function
+ * @param reduce_fnc reducer function
+ * @param acc_fnc optional accumulator function
+ * @param zero optional the zero value for the accumulator
+ * @param X left data set
+ * @param Y right data set
+ * @param num_reduces number of reducers
+ * @param stop_counter optional counter used in repeat operation
+ * @param orderp does the result need to be ordered?
+ * @return a new data source that contains the result
+ */
+ public final static DataSet mapReduce2 ( Tree mx, // left mapper function
+ Tree my, // right mapper function
+ Tree combine_fnc, // optional in-mapper combiner function
+ Tree reduce_fnc, // reducer function
+ Tree acc_fnc, // optional accumulator function
+ Tree zero, // optional the zero value for the accumulator
+ DataSet X, // left data set
+ DataSet Y, // right data set
+ int num_reduces, // number of reducers
+ String stop_counter, // optional counter used in repeat operation
+ boolean orderp ) // does the result need to be ordered?
+ throws Exception {
+ String newpath = new_path(conf);
+ conf.set("mrql.mapper.left",mx.toString());
+ conf.set("mrql.mapper.right",my.toString());
+ if (combine_fnc != null)
+ conf.set("mrql.combiner",combine_fnc.toString());
+ conf.set("mrql.reducer",reduce_fnc.toString());
+ if (zero != null) {
+ conf.set("mrql.accumulator",acc_fnc.toString());
+ conf.set("mrql.zero",zero.toString());
+ // the in-mapper combiner likes large data splits
+ conf.set("mapred.min.split.size","268435456"); // 256 MBs
+ } else conf.set("mrql.zero","");
+ conf.set("mrql.counter",stop_counter);
+ Job job = new Job(conf,newpath);
+ distribute_compiled_arguments(job.getConfiguration());
+ job.setMapOutputKeyClass(JoinKey.class);
+ job.setJarByClass(MapReducePlan.class);
+ job.setOutputKeyClass(MRContainer.class);
+ job.setOutputValueClass(MRContainer.class);
+ job.setPartitionerClass(MRContainerJoinPartitioner.class);
+ job.setSortComparatorClass(MRContainerSortComparator.class);
+ job.setGroupingComparatorClass(MRContainerGroupingComparator.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ FileOutputFormat.setOutputPath(job,new Path(newpath));
+ for (DataSource p: X.source)
+ MultipleInputs.addInputPath(job,new Path(p.path),(Class<? extends MapReduceMRQLFileInputFormat>)p.inputFormat,MapperLeft.class);
+ for (DataSource p: Y.source)
+ MultipleInputs.addInputPath(job,new Path(p.path),(Class<? extends MapReduceMRQLFileInputFormat>)p.inputFormat,MapperRight.class);
+ if (Config.trace && PlanGeneration.streamed_MapReduce2_reducer(reduce_fnc))
+ System.err.println("*** Streamed MapReduce2 reducer");
+ job.setReducerClass(JoinReducer.class);
+ if (num_reduces > 0)
+ job.setNumReduceTasks(num_reduces);
+ job.waitForCompletion(true);
+ long c = (stop_counter.equals("-")) ? 0
+ : job.getCounters().findCounter("mrql",stop_counter).getValue();
+ DataSource s = new BinaryDataSource(newpath,conf);
+ s.to_be_merged = orderp;
+ return new DataSet(s,c,outputRecords(job));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapJoinOperation.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapJoinOperation.java b/mapreduce/src/main/java/org/apache/mrql/MapJoinOperation.java
new file mode 100644
index 0000000..3ac4efa
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapJoinOperation.java
@@ -0,0 +1,213 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.io.*;
+import java.net.URI;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Enumeration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+import org.apache.hadoop.filecache.DistributedCache;
+
+
+/** The fragment-replicate join (map-side join) physical operator */
+final public class MapJoinOperation extends MapReducePlan {
+
+ /** the mapper of the MapJoin */
+ private final static class mapJoinMapper extends Mapper<MRContainer,MRContainer,MRContainer,MRContainer> {
+ private static String counter; // a Hadoop user-defined counter used in the repeat operation
+ private static Function reduce_fnc; // the reduce function
+ private static Function probe_map_fnc;
+ private static Hashtable<MRData,Bag> built_table;
+ private static Function acc_fnc; // aggregator
+ private static MRData result; // aggregation result
+ private static Tuple pair = new Tuple(2);
+ private static MRContainer container = new MRContainer(new MR_int(0));
+ private static Bag empty_bag = new Bag();
+ private static boolean mapJoinReduce = false;
+
+ private void write ( MRContainer key, MRData value, Context context )
+ throws IOException, InterruptedException {
+ if (result != null) { // aggregation
+ pair.set(0,result);
+ pair.set(1,value);
+ result = acc_fnc.eval(pair);
+ } else if (counter.equals("-")) {
+ container.set(value);
+ context.write(key,container);
+ } else { // increment the repetition counter if the repeat condition is true
+ Tuple t = (Tuple)value;
+ if (((MR_bool)t.second()).get())
+ context.getCounter("mrql",counter).increment(1);
+ container.set(t.first());
+ context.write(key,container);
+ }
+ }
+
+ @Override
+ public void map ( MRContainer key, MRContainer value, Context context )
+ throws IOException, InterruptedException {
+ for ( MRData e: (Bag)probe_map_fnc.eval(value.data()) ) {
+ Tuple p = (Tuple)e;
+ MRData pd = built_table.get(p.first());
+ if (pd == null)
+ pd = empty_bag;
+ pair.set(0,p.second());
+ pair.set(1,pd);
+ for ( MRData v: (Bag)reduce_fnc.eval(pair) )
+ write(key,v,context);
+ }
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ if (conf.get("mrql.mapJoinReduce") != null)
+ mapJoinReduce = true;
+ Tree code = Tree.parse(conf.get("mrql.inMap.reducer"));
+ reduce_fnc = functional_argument(conf,code);
+ code = Tree.parse(conf.get("mrql.probe_mapper"));
+ probe_map_fnc = functional_argument(conf,code);
+ if (!mapJoinReduce && conf.get("mrql.zero") != null) {
+ code = Tree.parse(conf.get("mrql.zero"));
+ result = Interpreter.evalE(code);
+ code = Tree.parse(conf.get("mrql.accumulator"));
+ acc_fnc = functional_argument(conf,code);
+ } else result = null;
+ counter = conf.get("mrql.counter");
+ built_table = new Hashtable<MRData,Bag>(Config.map_cache_size);
+ Bag res = new Bag();
+ URI[] uris = DistributedCache.getCacheFiles(conf);
+ Path[] local_paths = DistributedCache.getLocalCacheFiles(conf);
+ final FileSystem fs = FileSystem.getLocal(conf);
+ final Configuration fconf = conf;
+ for ( int i = 0; i < local_paths.length; i++ ) {
+ // hadoop 0.20.2 distributed cache doesn't work in stand-alone
+ final Path path = (conf.get("mapred.job.tracker").equals("local"))
+ ? new Path(uris[i].toString())
+ : local_paths[i];
+ if (path.getName().endsWith(".jar"))
+ continue;
+ res = res.union(new Bag(new BagIterator () {
+ final SequenceFile.Reader reader = new SequenceFile.Reader(fs,path,fconf);
+ final MRContainer key = new MRContainer(new MR_int(0));
+ final MRContainer value = new MRContainer(new MR_int(0));
+ public boolean hasNext () {
+ try {
+ boolean done = reader.next(key,value);
+ if (!done)
+ reader.close();
+ return done;
+ } catch (IOException e) {
+ throw new Error("Cannot collect values from distributed cache");
+ }
+ }
+ public MRData next () {
+ return value.data();
+ }
+ }));
+ };
+ for ( MRData e: res ) {
+ Tuple p = (Tuple)e;
+ Bag entries = built_table.get(p.first());
+ built_table.put(p.first(),
+ (entries == null)
+ ? (new Bag(p.second()))
+ : entries.add_element(p.second()));
+ }
+ } catch (Exception e) {
+ throw new Error("Cannot setup the mapJoin: "+e);
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ if (result != null) // emit the result of aggregation
+ context.write(new MRContainer(new MR_int(0)),new MRContainer(result));
+ built_table = null; // garbage-collect it
+ super.cleanup(context);
+ }
+ }
+
+ /** The fragment-replicate join (map-side join) physical operator
+ * @param probe_map_fnc left mapper function
+ * @param built_map_fnc right mapper function
+ * @param reduce_fnc reducer function
+ * @param acc_fnc optional accumulator function
+ * @param zero optional the zero value for the accumulator
+ * @param probe_dataset the map source
+ * @param built_dataset stored in distributed cache
+ * @param stop_counter optional counter used in repeat operation
+ * @return a new data source that contains the result
+ */
+ public final static DataSet mapJoin ( Tree probe_map_fnc, // left mapper function
+ Tree built_map_fnc, // right mapper function
+ Tree reduce_fnc, // reducer function
+ Tree acc_fnc, // optional accumulator function
+ Tree zero, // optional the zero value for the accumulator
+ DataSet probe_dataset, // the map source
+ DataSet built_dataset, // stored in distributed cache
+ String stop_counter ) // optional counter used in repeat operation
+ throws Exception {
+ DataSet ds = MapOperation.cMap(built_map_fnc,null,null,built_dataset,"-");
+ String newpath = new_path(conf);
+ conf.set("mrql.inMap.reducer",reduce_fnc.toString());
+ conf.set("mrql.probe_mapper",probe_map_fnc.toString());
+ conf.set("mrql.counter",stop_counter);
+ if (zero != null) {
+ conf.set("mrql.accumulator",acc_fnc.toString());
+ conf.set("mrql.zero",zero.toString());
+ conf.set("mapred.min.split.size","268435456");
+ } else conf.set("mrql.zero","");
+ Job job = new Job(conf,newpath);
+ distribute_compiled_arguments(job.getConfiguration());
+ job.setJarByClass(MapReducePlan.class);
+ job.setOutputKeyClass(MRContainer.class);
+ job.setOutputValueClass(MRContainer.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ PathFilter pf = new PathFilter () { public boolean accept ( Path path ) {
+ return !path.getName().startsWith("_");
+ } };
+ for (DataSource p: ds.source) { // distribute the built dataset
+ Path path = new Path(p.path);
+ for ( FileStatus s: path.getFileSystem(conf).listStatus(path,pf) )
+ DistributedCache.addCacheFile(s.getPath().toUri(),job.getConfiguration());
+ };
+ for (DataSource p: probe_dataset.source)
+ MultipleInputs.addInputPath(job,new Path(p.path),(Class<? extends MapReduceMRQLFileInputFormat>)p.inputFormat,mapJoinMapper.class);
+ FileOutputFormat.setOutputPath(job,new Path(newpath));
+ job.setNumReduceTasks(0);
+ job.waitForCompletion(true);
+ long c = (stop_counter.equals("-")) ? 0
+ : job.getCounters().findCounter("mrql",stop_counter).getValue();
+ return new DataSet(new BinaryDataSource(newpath,conf),c,outputRecords(job));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapOperation.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapOperation.java b/mapreduce/src/main/java/org/apache/mrql/MapOperation.java
new file mode 100644
index 0000000..b5c9a8f
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapOperation.java
@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.io.*;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+
+/** The MapReduce operation that use a map stage only */
+final public class MapOperation extends MapReducePlan {
+
+ /** The mapper of Map */
+ private final static class cMapMapper extends Mapper<MRContainer,MRContainer,MRContainer,MRContainer> {
+ private static String counter; // a Hadoop user-defined counter used in the repeat operation
+ private static Function map_fnc; // the mapper function
+ private static Function acc_fnc; // aggregator
+ private static MRData result; // aggregation result
+ private static Tuple pair = new Tuple(2);
+ private static MRContainer container = new MRContainer(new MR_int(0));
+
+ private void write ( MRContainer key, MRData value, Context context )
+ throws IOException, InterruptedException {
+ if (result != null) { // aggregation
+ pair.set(0,result);
+ pair.set(1,value);
+ result = acc_fnc.eval(pair);
+ } else if (counter.equals("-")) {
+ container.set(value);
+ context.write(key,container);
+ } else { // increment the repetition counter if the repeat condition is true
+ Tuple t = (Tuple)value;
+ if (((MR_bool)t.second()).get())
+ context.getCounter("mrql",counter).increment(1);
+ container.set(t.first());
+ context.write(key,container);
+ }
+ }
+
+ @Override
+ public void map ( MRContainer key, MRContainer value, Context context )
+ throws IOException, InterruptedException {
+ for ( MRData e: (Bag)map_fnc.eval(value.data()) )
+ write(key,e,context);
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ Tree code = Tree.parse(conf.get("mrql.mapper"));
+ map_fnc = functional_argument(conf,code);
+ if (conf.get("mrql.zero") != null) {
+ code = Tree.parse(conf.get("mrql.zero"));
+ result = Interpreter.evalE(code);
+ code = Tree.parse(conf.get("mrql.accumulator"));
+ acc_fnc = functional_argument(conf,code);
+ } else result = null;
+ counter = conf.get("mrql.counter");
+ } catch (Exception e) {
+ throw new Error("Cannot retrieve the mapper plan");
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ if (result != null) // emit the result of aggregation
+ context.write(new MRContainer(new MR_int(0)),new MRContainer(result));
+ super.cleanup(context);
+ }
+ }
+
+ /** The cMap physical operator
+ * @param map_fnc mapper function
+ * @param acc_fnc optional accumulator function
+ * @param zero optional the zero value for the accumulator
+ * @param source input data source
+ * @param stop_counter optional counter used in repeat operation
+ * @return a new data source that contains the result
+ */
+ public final static DataSet cMap ( Tree map_fnc, // mapper function
+ Tree acc_fnc, // optional accumulator function
+ Tree zero, // optional the zero value for the accumulator
+ DataSet source, // input data source
+ String stop_counter ) // optional counter used in repeat operation
+ throws Exception {
+ String newpath = new_path(conf);
+ conf.set("mrql.mapper",map_fnc.toString());
+ conf.set("mrql.counter",stop_counter);
+ if (zero != null) {
+ conf.set("mrql.accumulator",acc_fnc.toString());
+ conf.set("mrql.zero",zero.toString());
+ conf.set("mapred.min.split.size","268435456");
+ } else conf.set("mrql.zero","");
+ Job job = new Job(conf,newpath);
+ distribute_compiled_arguments(job.getConfiguration());
+ job.setJarByClass(MapReducePlan.class);
+ job.setOutputKeyClass(MRContainer.class);
+ job.setOutputValueClass(MRContainer.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ for (DataSource p: source.source)
+ MultipleInputs.addInputPath(job,new Path(p.path),(Class<? extends MapReduceMRQLFileInputFormat>)p.inputFormat,cMapMapper.class);
+ FileOutputFormat.setOutputPath(job,new Path(newpath));
+ job.setNumReduceTasks(0);
+ job.waitForCompletion(true);
+ long c = (stop_counter.equals("-")) ? 0
+ : job.getCounters().findCounter("mrql",stop_counter).getValue();
+ return new DataSet(new BinaryDataSource(newpath,conf),c,outputRecords(job));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapReduceBinaryInputFormat.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceBinaryInputFormat.java b/mapreduce/src/main/java/org/apache/mrql/MapReduceBinaryInputFormat.java
new file mode 100644
index 0000000..36ad094
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceBinaryInputFormat.java
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import java.io.*;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
+
+
+/** Input format for Apache Hadoop sequence files */
+final public class MapReduceBinaryInputFormat extends MapReduceMRQLFileInputFormat {
+ final static SequenceFileInputFormat<MRContainer,MRContainer> inputFormat
+ = new SequenceFileInputFormat<MRContainer,MRContainer>();
+
+ public RecordReader<MRContainer,MRContainer>
+ createRecordReader ( InputSplit split,
+ TaskAttemptContext context ) throws IOException, InterruptedException {
+ return inputFormat.createRecordReader(split,context);
+ }
+
+ /** collect the data from multiple sequence files at the path directory into a Bag
+ * @param path the path directory
+ * @return a Bag that contains all data
+ */
+ public Bag materialize ( final Path path ) throws IOException {
+ final FileSystem fs = path.getFileSystem(Plan.conf);
+ final FileStatus[] ds
+ = fs.listStatus(path,
+ new PathFilter () {
+ public boolean accept ( Path path ) {
+ return !path.getName().startsWith("_");
+ }
+ });
+ if (ds.length > 0)
+ return new Bag(new BagIterator () {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs,ds[0].getPath(),Plan.conf);
+ MRContainer key = new MRContainer(new MR_int(0));
+ MRContainer value = new MRContainer(new MR_int(0));
+ int i = 1;
+ public boolean hasNext () {
+ try {
+ if (reader.next(key,value))
+ return true;
+ do {
+ if (i >= ds.length)
+ return false;
+ reader.close();
+ reader = new SequenceFile.Reader(fs,ds[i++].getPath(),Plan.conf);
+ } while (!reader.next(key,value));
+ return true;
+ } catch (IOException e) {
+ throw new Error("Cannot collect values from an intermediate result");
+ }
+ }
+ public MRData next () {
+ return value.data();
+ }
+ });
+ return new Bag();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen b/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
new file mode 100644
index 0000000..e733715
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceEvaluator.gen
@@ -0,0 +1,296 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import java_cup.runtime.*;
+import org.apache.mrql.gen.*;
+import java.util.*;
+import java.io.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+
+
+/** Evaluates physical plans in Apache Hadoop MapReduce mode */
+final public class MapReduceEvaluator extends Evaluator {
+
+ /** initialize the MapReduce evaluator */
+ final public void init ( Configuration conf ) {
+ Config.map_reduce_mode = true;
+ if (Config.hadoop_mode)
+ if (Config.local_mode) {
+ conf.set("mapred.job.tracker","local");
+ conf.set("fs.default.name","file:///");
+ } else {
+ conf.set("mapred.job.tracker",System.getenv("MAPRED_JOB_TRACKER"));
+ conf.set("fs.default.name",System.getenv("FS_DEFAULT_NAME"));
+ }
+ }
+
+ /** shutdown the MapReduce evaluator */
+ final public void shutdown ( Configuration conf ) {
+ }
+
+ /** initialize the query evaluation */
+ final public void initialize_query () {
+ }
+
+ /** create a new evaluation configuration */
+ final public Configuration new_configuration () {
+ return new Configuration();
+ }
+
+ /** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
+ final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
+ return MapReduceParsedInputFormat.class;
+ }
+
+ /** return the FileInputFormat for binary files */
+ final public Class<? extends MRQLFileInputFormat> binaryInputFormat () {
+ return MapReduceBinaryInputFormat.class;
+ }
+
+ /** return the FileInputFormat for data generator files */
+ final public Class<? extends MRQLFileInputFormat> generatorInputFormat () {
+ return MapReduceGeneratorInputFormat.class;
+ }
+
+ /** The Aggregate physical operator
+ * @param acc_fnc the accumulator function from (T,T) to T
+ * @param zero the zero element of type T
+ * @param plan the plan that constructs the dataset that contains the bag of values {T}
+ * @param env contains bindings fro variables to values (MRData)
+ * @return the aggregation result of type T
+ */
+ final public MRData aggregate ( Tree acc_fnc,
+ Tree zero,
+ Tree plan,
+ Environment env ) throws Exception {
+ return MapReducePlan.aggregate(acc_fnc,zero,eval(plan,env,"-"));
+ }
+
+ /** Evaluate a loop a fixed # of times */
+ final public Tuple loop ( Tree e, Environment env ) throws Exception {
+ match e {
+ case Loop(lambda(tuple(...vs),tuple(...bs)),tuple(...ss),`num):
+ int limit = ((MR_int)evalE(num,env)).get();
+ MR_dataset[] s = new MR_dataset[vs.length()];
+ for ( int i = 0; i < vs.length(); i++ )
+ s[i] = new MR_dataset(eval(ss.nth(i),env,"-"));
+ for ( int n = 0; n < limit; n++ ) {
+ Environment nenv = env;
+ for ( int i = 0; i < vs.length(); i ++ )
+ nenv = new Environment(vs.nth(i).toString(),s[i],nenv);
+ for ( int i = 0; i < vs.length(); i ++ )
+ s[i] = new MR_dataset(eval(bs.nth(i),nenv,"-"));
+ };
+ return new Tuple(s);
+ };
+ throw new Error("Wrong Loop format");
+ }
+
+ /** Evaluate an MRQL physical plan using Hadoop and print tracing info
+ * @param e the physical plan
+ * @param env contains bindings fro variables to values (MRData)
+ * @return a DataSet (stored in HDFS)
+ */
+ final public DataSet eval ( final Tree e,
+ final Environment env,
+ final String counter ) {
+ if (Config.trace_execution) {
+ tab_count += 3;
+ System.out.println(tabs(tab_count)+print_query(e));
+ };
+ DataSet res = evalD(e,env,counter);
+ if (Config.trace_execution)
+ try {
+ 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);
+ };
+ return res;
+ }
+
+ /** Evaluate MRQL physical operators using Hadoop
+ * @param e the physical plan
+ * @param env contains bindings fro variables to values (MRData)
+ * @return a DataSet (stored in HDFS)
+ */
+ final DataSet evalD ( final Tree e,
+ final Environment env,
+ final String counter ) {
+ try {
+ match e {
+ case cMap(`f,`s):
+ return MapOperation.cMap(closure(f,env),null,null,eval(s,env,"-"),counter);
+ case AggregateMap(`f,`acc,`zero,`s):
+ return MapOperation.cMap(closure(f,env),closure(acc,env),
+ (zero.equals(#<null>))?null:zero,
+ eval(s,env,"-"),counter);
+ case MapReduce(`m,`r,`s,`o):
+ return MapReduceOperation.mapReduce(closure(m,env),#<null>,closure(r,env),
+ null,null,
+ eval(s,env,"-"),
+ Config.nodes,counter,
+ o.equals(#<true>));
+ case MapAggregateReduce(`m,`r,`acc,`zero,`s,`o):
+ return MapReduceOperation.mapReduce(closure(m,env),null,closure(r,env),
+ closure(acc,env),
+ (zero.equals(#<null>))?null:zero,
+ eval(s,env,"-"),
+ Config.nodes,counter,
+ o.equals(#<true>));
+ case MapCombineReduce(`m,`c,`r,`s,`o):
+ return MapReduceOperation.mapReduce(closure(m,env),closure(c,env),closure(r,env),
+ null,null,
+ eval(s,env,"-"),
+ Config.nodes,counter,
+ o.equals(#<true>));
+ case CrossProduct(`mx,`my,`r,`x,`y):
+ return CrossProductOperation.crossProduct(closure(mx,env),closure(my,env),closure(r,env),
+ null,null,
+ eval(x,env,"-"),
+ eval(y,env,"-"),
+ counter);
+ case CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y):
+ return CrossProductOperation.crossProduct(closure(mx,env),closure(my,env),closure(r,env),
+ closure(acc,env),
+ (zero.equals(#<null>))?null:zero,
+ eval(x,env,"-"),
+ eval(y,env,"-"),
+ counter);
+ case MapReduce2(`mx,`my,`r,`x,`y,`o):
+ return eval(#<MapAggregateReduce2(`mx,`my,`r,null,null,`x,`y,`o)>,env,counter);
+ case MapCombineReduce2(`mx,`my,`c,`r,`x,`y,`o):
+ return JoinOperation.mapReduce2(closure(mx,env),closure(my,env),
+ closure(c,env),
+ closure(r,env),
+ null,null,
+ eval(x,env,"-"),
+ eval(y,env,"-"),
+ Config.nodes,counter,
+ o.equals(#<true>));
+ case MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,`o):
+ return JoinOperation.mapReduce2(closure(mx,env),closure(my,env),
+ null,
+ closure(r,env),closure(acc,env),
+ (zero.equals(#<null>))?null:zero,
+ eval(x,env,"-"),
+ eval(y,env,"-"),
+ Config.nodes,counter,
+ o.equals(#<true>));
+ 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.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),
+ X,Y,
+ Config.nodes,
+ n,m,counter);
+ case MapJoin(`mx,`my,`r,`x,`y):
+ return MapJoinOperation.mapJoin(closure(mx,env),closure(my,env),closure(r,env),
+ null,null,
+ eval(x,env,"-"),
+ eval(y,env,"-"),
+ counter);
+ case MapAggregateJoin(`mx,`my,`r,`acc,`zero,`x,`y):
+ return MapJoinOperation.mapJoin(closure(mx,env),closure(my,env),closure(r,env),
+ closure(acc,env),
+ (zero.equals(#<null>))?null:zero,
+ eval(x,env,"-"),
+ eval(y,env,"-"),
+ counter);
+ case BinarySource(`file,_):
+ return Plan.binarySource(file.stringValue());
+ case ParsedSource(`parser,`file,...args):
+ Class<? extends Parser> p = DataSource.parserDirectory.get(parser.toString());
+ if (p == null)
+ throw new Error("Unknown parser: "+parser);
+ return Plan.parsedSource(p,((MR_string)evalE(file,env)).get(),args);
+ case Merge(`x,`y):
+ return Plan.merge(eval(x,env,"-"),eval(y,env,"-"));
+ case Repeat(lambda(`v,`b),`s,`n):
+ final String nm = v.toString();
+ final Tree body = b;
+ Function loop_fnc = new Function () {
+ public MRData eval ( MRData s ) {
+ return new MR_dataset(evaluator.eval(body,new Environment(nm,s,env),nm));
+ }; };
+ return MapReducePlan.repeat(loop_fnc,eval(s,env,"-"),((MR_int)evalE(n,env)).get());
+ case Closure(lambda(`v,`b),`s,`n):
+ final String nm = v.toString();
+ final Tree body = b;
+ Function loop_fnc = new Function () {
+ public MRData eval ( MRData s ) {
+ return new MR_dataset(evaluator.eval(body,new Environment(nm,s,env),"-"));
+ }; };
+ return MapReducePlan.closure(loop_fnc,eval(s,env,"-"),((MR_int)evalE(n,env)).get());
+ case Generator(`min,`max,`size):
+ return Plan.generator(((MR_long)evalE(min,env)).get(),
+ ((MR_long)evalE(max,env)).get(),
+ ((MR_long)evalE(size,env)).get());
+ case let(`v,`u,`body):
+ return eval(body,new Environment(v.toString(),evalE(u,env),env),"-");
+ case Let(`v,`u,`body):
+ return eval(body,new Environment(v.toString(),new MR_dataset(eval(u,env,"-")),env),"-");
+ case If(`c,`x,`y):
+ if (((MR_bool)evalE(c,env)).get())
+ return eval(x,env,"-");
+ else return eval(y,env,"-");
+ case apply(`f,`arg):
+ if (!f.is_variable())
+ return ((MR_dataset)evalF(f,env).eval(evalE(arg))).dataset();
+ MRData fnc = variable_lookup(f.toString(),global_env);
+ if (fnc == null)
+ throw new Error("Unknown function: "+f);
+ MRData t = evalE(arg,env);
+ if (!(t instanceof Tuple))
+ throw new Error("Expected a tuple in function application: "+t);
+ return ((MR_dataset)((Lambda)fnc).lambda().eval(t)).dataset();
+ case `v:
+ if (!v.is_variable())
+ fail;
+ MRData x = variable_lookup(v.toString(),env);
+ if (x != null)
+ if (x instanceof MR_dataset)
+ return ((MR_dataset)x).dataset();
+ x = variable_lookup(v.toString(),global_env);
+ if (x != null)
+ if (x instanceof MR_dataset)
+ return ((MR_dataset)x).dataset();
+ throw new Error("Variable "+v+" is not bound");
+ };
+ throw new Error("Cannot evaluate the map-reduce plan: "+e);
+ } catch (Error msg) {
+ if (!Config.trace)
+ throw new Error(msg.getMessage());
+ System.err.println(msg.getMessage());
+ throw new Error("Evaluation error in: "+print_query(e));
+ } catch (Exception ex) {
+ System.err.println(ex.getMessage());
+ ex.printStackTrace();
+ throw new Error("Evaluation error in: "+print_query(e));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapReduceGeneratorInputFormat.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceGeneratorInputFormat.java b/mapreduce/src/main/java/org/apache/mrql/MapReduceGeneratorInputFormat.java
new file mode 100644
index 0000000..9a8ec55
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceGeneratorInputFormat.java
@@ -0,0 +1,114 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import java.io.*;
+import java.util.Iterator;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+
+
+/** the FileInputFormat for data generators: it creates HDFS files, where each file contains
+ * an (offset,size) pair that generates the range of values [offset,offset+size] */
+final public class MapReduceGeneratorInputFormat extends MapReduceMRQLFileInputFormat {
+ public static class GeneratorRecordReader extends RecordReader<MRContainer,MRContainer> {
+ final long offset;
+ final long size;
+ long index;
+ SequenceFile.Reader reader;
+
+ public GeneratorRecordReader ( FileSplit split,
+ TaskAttemptContext context ) throws IOException {
+ Configuration conf = context.getConfiguration();
+ Path path = split.getPath();
+ FileSystem fs = path.getFileSystem(conf);
+ reader = new SequenceFile.Reader(path.getFileSystem(conf),path,conf);
+ MRContainer key = new MRContainer();
+ MRContainer value = new MRContainer();
+ reader.next(key,value);
+ offset = ((MR_long)((Tuple)(value.data())).first()).get();
+ size = ((MR_long)((Tuple)(value.data())).second()).get();
+ index = -1;
+ }
+
+ public boolean nextKeyValue () throws IOException {
+ index++;
+ return index < size;
+ }
+
+ public MRContainer getCurrentKey () throws IOException {
+ return new MRContainer(new MR_long(index));
+ }
+
+ public MRContainer getCurrentValue () throws IOException {
+ return new MRContainer(new MR_long(offset+index));
+ }
+
+ public void close () throws IOException { reader.close(); }
+
+ public float getProgress () throws IOException {
+ return index / (float)size;
+ }
+
+ public void initialize ( InputSplit split, TaskAttemptContext context ) throws IOException { }
+ }
+
+ public RecordReader<MRContainer,MRContainer>
+ createRecordReader ( InputSplit split, TaskAttemptContext context ) throws IOException {
+ return new GeneratorRecordReader((FileSplit)split,context);
+ }
+
+ /** Insert all results from the generators stored in path into a Bag.
+ * The Bag is lazily constructed.
+ * @param path the path directory that contains the generator data (offset,size)
+ * @return a Bag that contains all data
+ */
+ public Bag materialize ( final Path path ) throws IOException {
+ Configuration conf = Plan.conf;
+ FileSystem fs = path.getFileSystem(conf);
+ final SequenceFile.Reader reader = new SequenceFile.Reader(path.getFileSystem(conf),path,conf);
+ final MRContainer key = new MRContainer();
+ final MRContainer value = new MRContainer();
+ return new Bag(new BagIterator () {
+ long offset = 0;
+ long size = 0;
+ long i = 0;
+ public boolean hasNext () {
+ if (++i >= offset+size)
+ try {
+ if (!reader.next(key,value))
+ return false;
+ offset = ((MR_long)((Tuple)(value.data())).first()).get();
+ size = ((MR_long)((Tuple)(value.data())).second()).get();
+ i = offset;
+ } catch (IOException e) {
+ throw new Error("Cannot collect values from a generator");
+ };
+ return true;
+ }
+ public MRData next () {
+ return new MR_long(i);
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapReduceMRQLFileInputFormat.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceMRQLFileInputFormat.java b/mapreduce/src/main/java/org/apache/mrql/MapReduceMRQLFileInputFormat.java
new file mode 100644
index 0000000..d0760b4
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceMRQLFileInputFormat.java
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import java.io.*;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+
+
+/** A superclass for all MRQL FileInputFormats */
+abstract public class MapReduceMRQLFileInputFormat extends FileInputFormat<MRContainer,MRContainer> implements MRQLFileInputFormat {
+ public MapReduceMRQLFileInputFormat () {}
+
+ /** record reader for map-reduce */
+ abstract public RecordReader<MRContainer,MRContainer>
+ createRecordReader ( InputSplit split,
+ TaskAttemptContext context ) throws IOException, InterruptedException;
+
+ /** materialize the input file into a memory Bag */
+ abstract public Bag materialize ( final Path path ) throws IOException;
+
+ /** 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
+ */
+ public final Bag collect ( final DataSet x, boolean strip ) throws Exception {
+ Bag res = new Bag();
+ for ( DataSource s: x.source )
+ 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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapReduceOperation.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceOperation.java b/mapreduce/src/main/java/org/apache/mrql/MapReduceOperation.java
new file mode 100644
index 0000000..39864b3
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceOperation.java
@@ -0,0 +1,273 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.io.*;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Enumeration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+
+
+/** The MapReduce operation that uses an in-mapper combiner to partially reduce groups during mapping */
+final public class MapReduceOperation extends MapReducePlan {
+
+ public final static class MRContainerPartitioner extends Partitioner<MRContainer,MRContainer> {
+ final public int getPartition ( MRContainer key, MRContainer value, int numPartitions ) {
+ return Math.abs(key.hashCode()) % numPartitions;
+ }
+ }
+
+ /** The mapper of the MapReduce operation */
+ private final static class MRMapper extends Mapper<MRContainer,MRContainer,MRContainer,MRContainer> {
+ private static Function map_fnc; // the map function
+ private static Function combine_fnc; // the combiner function
+ private static Hashtable<MRData,MRData> hashTable; // in-mapper combiner
+ private static int index;
+ private static Tuple tkey = new Tuple(2);
+ private static Bag tbag = new Bag(2);
+ private static MRContainer ckey = new MRContainer(new MR_int(0));
+ private static MRContainer cvalue = new MRContainer(new MR_int(0));
+
+ @Override
+ public void map ( MRContainer key, MRContainer value, Context context )
+ throws IOException, InterruptedException {
+ for ( MRData e: (Bag)map_fnc.eval(value.data()) ) {
+ Tuple p = (Tuple)e;
+ if (hashTable == null) {
+ ckey.set(p.first());
+ cvalue.set(p.second());
+ context.write(ckey,cvalue);
+ } else {
+ // in-mapper combiner
+ MRData old = hashTable.get(p.first());
+ if (old == null) {
+ if (index++ == Config.map_cache_size)
+ flush_table(context);
+ hashTable.put(p.first(),p.second());
+ } else {
+ tkey.set(0,p.first());
+ tbag.clear();
+ tbag.add_element(p.second()).add_element(old);
+ tkey.set(1,tbag);
+ for ( MRData x: (Bag)combine_fnc.eval(tkey) )
+ hashTable.put(p.first(),x); // normally, done once
+ }
+ }
+ }
+ }
+
+ private static void flush_table ( Context context ) throws IOException, InterruptedException {
+ Enumeration<MRData> en = hashTable.keys();
+ while (en.hasMoreElements()) {
+ MRData key = en.nextElement();
+ ckey.set(key);
+ MRData value = hashTable.get(key);
+ cvalue.set(value);
+ if (value != null)
+ context.write(ckey,cvalue);
+ };
+ index = 0;
+ hashTable.clear();
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ Tree code = Tree.parse(conf.get("mrql.mapper"));
+ map_fnc = functional_argument(conf,code);
+ code = Tree.parse(conf.get("mrql.combiner"));
+ hashTable = null;
+ if (code != null && !code.equals(new VariableLeaf("null"))) {
+ combine_fnc = functional_argument(conf,code);
+ hashTable = new Hashtable<MRData,MRData>(Config.map_cache_size);
+ index = 0;
+ }
+ } catch (Exception e) {
+ throw new Error("Cannot retrieve the mapper plan");
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ if (hashTable != null)
+ flush_table(context);
+ hashTable = null; // garbage-collect it
+ super.cleanup(context);
+ }
+ }
+
+ /** The reducer of the MapReduce operation */
+ private final static class MRReducer extends Reducer<MRContainer,MRContainer,MRContainer,MRContainer> {
+ private static String counter; // a Hadoop user-defined counter used in the repeat operation
+ private static Function reduce_fnc; // the reduce function
+ private static Bag s = new Bag(); // a cached bag of input fragments
+ private static Function acc_fnc; // aggregator
+ private static MRData result; // aggregation result
+ private static boolean streamed = false;
+ private static Tuple pair = new Tuple(2);
+ private static MRContainer container = new MRContainer(new MR_int(0));
+
+ private void write ( MRContainer key, MRData value, Context context )
+ throws IOException, InterruptedException {
+ if (result != null) { // aggregation
+ pair.set(0,result);
+ pair.set(1,value);
+ result = acc_fnc.eval(pair);
+ } else if (counter.equals("-")) {
+ container.set(value);
+ context.write(key,container);
+ } else { // increment the repetition counter if the repeat condition is true
+ Tuple t = (Tuple)value;
+ if (((MR_bool)t.second()).get())
+ context.getCounter("mrql",counter).increment(1);
+ container.set(t.first());
+ context.write(key,container);
+ }
+ }
+
+ @Override
+ public void reduce ( MRContainer key, Iterable<MRContainer> values, Context context )
+ throws IOException, InterruptedException {
+ if (!streamed) { // store the values in a Bag and then reduce
+ s.clear();
+ for ( MRContainer val: values )
+ s.add(val.data());
+ pair.set(0,key.data());
+ pair.set(1,s);
+ for ( MRData e: (Bag)reduce_fnc.eval(pair) )
+ write(key,e,context);
+ } else { // it accesses the values in stream-like fashion
+ final Iterator<MRContainer> iterator = values.iterator();
+ Bag s = new Bag(new BagIterator() {
+ public boolean hasNext () {
+ return iterator.hasNext();
+ }
+ public MRData next () {
+ return iterator.next().data();
+ }
+ });
+ pair.set(0,key.data());
+ pair.set(1,s);
+ for ( MRData e: (Bag)reduce_fnc.eval(pair) )
+ write(key,e,context);
+ }
+ }
+
+ @Override
+ protected void setup ( Context context ) throws IOException,InterruptedException {
+ super.setup(context);
+ try {
+ Configuration conf = context.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ Tree code = Tree.parse(conf.get("mrql.reducer"));
+ reduce_fnc = functional_argument(conf,code);
+ streamed = PlanGeneration.streamed_MapReduce_reducer(code);
+ if (conf.get("mrql.zero") != null) {
+ code = Tree.parse(conf.get("mrql.zero"));
+ result = Interpreter.evalE(code);
+ code = Tree.parse(conf.get("mrql.accumulator"));
+ acc_fnc = functional_argument(conf,code);
+ } else result = null;
+ counter = conf.get("mrql.counter");
+ } catch (Exception e) {
+ throw new Error("Cannot retrieve the reduce plan");
+ }
+ }
+
+ @Override
+ protected void cleanup ( Context context ) throws IOException,InterruptedException {
+ if (result != null) // emit the result of aggregation
+ context.write(new MRContainer(new MR_int(0)),new MRContainer(result));
+ super.cleanup(context);
+ }
+ }
+
+ /**
+ * The MapReduce physical operator
+ * @param map_fnc the mapper function
+ * @param combine_fnc optional in-mapper combiner function
+ * @param reduce_fnc the reducer function
+ * @param acc_fnc optional accumulator function
+ * @param zero optional the zero value for the accumulator
+ * @param source the input data source
+ * @param num_reduces number of reducers
+ * @param stop_counter optional counter used in repeat operation
+ * @param orderp does the result need to be ordered?
+ * @return a new data source that contains the result
+ */
+ public final static DataSet mapReduce ( Tree map_fnc, // mapper function
+ Tree combine_fnc, // optional in-mapper combiner function
+ Tree reduce_fnc, // reducer function
+ Tree acc_fnc, // optional accumulator function
+ Tree zero, // optional the zero value for the accumulator
+ DataSet source, // input data source
+ int num_reduces, // number of reducers
+ String stop_counter, // optional counter used in repeat operation
+ boolean orderp ) // does the result need to be ordered?
+ throws Exception {
+ String newpath = new_path(conf);
+ conf.set("mrql.mapper",map_fnc.toString());
+ if (combine_fnc != null)
+ conf.set("mrql.combiner",combine_fnc.toString());
+ conf.set("mrql.reducer",reduce_fnc.toString());
+ if (zero != null) { // will use in-mapper combiner
+ conf.set("mrql.accumulator",acc_fnc.toString());
+ conf.set("mrql.zero",zero.toString());
+ // the in-mapper combiner likes large data splits
+ conf.set("mapred.min.split.size","268435456"); // 256 MBs
+ } else conf.set("mrql.zero","");
+ conf.set("mrql.counter",stop_counter);
+ Job job = new Job(conf,newpath);
+ distribute_compiled_arguments(job.getConfiguration());
+ job.setJarByClass(MapReducePlan.class);
+ job.setOutputKeyClass(MRContainer.class);
+ job.setOutputValueClass(MRContainer.class);
+ job.setPartitionerClass(MRContainerPartitioner.class);
+ job.setSortComparatorClass(MRContainerKeyComparator.class);
+ job.setGroupingComparatorClass(MRContainerKeyComparator.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ for (DataSource p: source.source)
+ MultipleInputs.addInputPath(job,new Path(p.path),(Class<? extends MapReduceMRQLFileInputFormat>)p.inputFormat,MRMapper.class);
+ FileOutputFormat.setOutputPath(job,new Path(newpath));
+ job.setReducerClass(MRReducer.class);
+ if (Config.trace && PlanGeneration.streamed_MapReduce_reducer(reduce_fnc))
+ System.err.println("*** Streamed MapReduce reducer");
+ if (num_reduces > 0)
+ job.setNumReduceTasks(num_reduces);
+ job.waitForCompletion(true);
+ long c = (stop_counter.equals("-")) ? 0
+ : job.getCounters().findCounter("mrql",stop_counter).getValue();
+ DataSource s = new BinaryDataSource(newpath,conf);
+ s.to_be_merged = orderp;
+ return new DataSet(s,c,outputRecords(job));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/mapreduce/src/main/java/org/apache/mrql/MapReduceParsedInputFormat.java
----------------------------------------------------------------------
diff --git a/mapreduce/src/main/java/org/apache/mrql/MapReduceParsedInputFormat.java b/mapreduce/src/main/java/org/apache/mrql/MapReduceParsedInputFormat.java
new file mode 100644
index 0000000..e588dbd
--- /dev/null
+++ b/mapreduce/src/main/java/org/apache/mrql/MapReduceParsedInputFormat.java
@@ -0,0 +1,139 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.Trees;
+import java.io.*;
+import java.util.Iterator;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+
+
+/** A FileInputFormat for text files (CVS, XML, JSON, ...) */
+final public class MapReduceParsedInputFormat extends MapReduceMRQLFileInputFormat {
+
+ public static class ParsedRecordReader extends RecordReader<MRContainer,MRContainer> {
+ final FSDataInputStream fsin;
+ final long start;
+ final long end;
+ Iterator<MRData> result;
+ MRData data;
+ Parser parser;
+
+ public ParsedRecordReader ( FileSplit split,
+ TaskAttemptContext context,
+ Class<? extends Parser> parser_class,
+ Trees args ) throws IOException {
+ Configuration conf = context.getConfiguration();
+ start = split.getStart();
+ end = start + split.getLength();
+ Path file = split.getPath();
+ FileSystem fs = file.getFileSystem(conf);
+ fsin = fs.open(split.getPath());
+ try {
+ parser = parser_class.newInstance();
+ } catch (Exception ex) {
+ throw new Error("Unrecognized parser:"+parser_class);
+ };
+ parser.initialize(args);
+ parser.open(fsin,start,end);
+ result = null;
+ }
+
+ public boolean nextKeyValue () throws IOException {
+ while (result == null || !result.hasNext()) {
+ String s = parser.slice();
+ if (s == null)
+ return false;
+ result = parser.parse(s).iterator();
+ };
+ data = (MRData)result.next();
+ return true;
+ }
+
+ public MRContainer getCurrentKey () throws IOException {
+ return new MRContainer(new MR_long(fsin.getPos()));
+ }
+
+ public MRContainer getCurrentValue () throws IOException {
+ return new MRContainer(data);
+ }
+
+ public long getPos () throws IOException { return fsin.getPos(); }
+
+ public void close () throws IOException { fsin.close(); }
+
+ public float getProgress () throws IOException {
+ if (end == start)
+ return 0.0f;
+ else return Math.min(1.0f, (getPos() - start) / (float)(end - start));
+ }
+
+ public void initialize ( InputSplit split, TaskAttemptContext context ) throws IOException { }
+ }
+
+ public RecordReader<MRContainer,MRContainer>
+ createRecordReader ( InputSplit split,
+ TaskAttemptContext context ) throws IOException, InterruptedException {
+ Configuration conf = context.getConfiguration();
+ String path = ((FileSplit)split).getPath().toString();
+ ParsedDataSource ds = (ParsedDataSource)DataSource.get(path,conf);
+ return new ParsedRecordReader((FileSplit)split,context,ds.parser,(Trees)ds.args);
+ }
+
+ /** Find the parser associated with each file in the path and parse the file,
+ * inserting all results into a Bag. The Bag is lazily constructed.
+ * @param path the path directory with the files
+ * @return a Bag that contains all data
+ */
+ public Bag materialize ( final Path path ) throws IOException {
+ Configuration conf = Plan.conf;
+ ParsedDataSource ds = (ParsedDataSource)DataSource.get(path.toString(),conf);
+ FileSystem fs = path.getFileSystem(conf);
+ FSDataInputStream fsin = fs.open(path);
+ Parser p;
+ try {
+ p = ds.parser.newInstance();
+ } catch (Exception ex) {
+ throw new Error("Unrecognized parser:"+ds.parser);
+ };
+ final Parser parser = p;
+ parser.initialize(ds.args);
+ parser.open(fsin,0,Long.MAX_VALUE);
+ return new Bag(new BagIterator () {
+ Iterator<MRData> iter;
+ public boolean hasNext () {
+ while (iter == null || !iter.hasNext()) {
+ String line = parser.slice();
+ if (line == null)
+ return false;
+ iter = parser.parse(line).iterator();
+ };
+ return true;
+ }
+ public MRData next () {
+ return iter.next();
+ }
+ });
+ }
+}