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/01 03:29:30 UTC
[3/5] MRQL-19: Fix license issues to prepare a new release candidate
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/MapReduce/MapReduceEvaluator.gen
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReduceEvaluator.gen b/src/main/java/MapReduce/MapReduceEvaluator.gen
new file mode 100644
index 0000000..3e53a92
--- /dev/null
+++ b/src/main/java/MapReduce/MapReduceEvaluator.gen
@@ -0,0 +1,300 @@
+/**
+ * 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();
+ }
+
+ /** 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;
+ }
+
+ /** 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)
+ +"-> "+MapReducePlan.collect(res));
+ 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.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");
+ 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/17a0ed95/src/main/java/MapReduce/MapReduceGeneratorInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReduceGeneratorInputFormat.java b/src/main/java/MapReduce/MapReduceGeneratorInputFormat.java
new file mode 100644
index 0000000..9a8ec55
--- /dev/null
+++ b/src/main/java/MapReduce/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/17a0ed95/src/main/java/MapReduce/MapReduceMRQLFileInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReduceMRQLFileInputFormat.java b/src/main/java/MapReduce/MapReduceMRQLFileInputFormat.java
new file mode 100644
index 0000000..d0760b4
--- /dev/null
+++ b/src/main/java/MapReduce/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/17a0ed95/src/main/java/MapReduce/MapReduceOperation.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReduceOperation.java b/src/main/java/MapReduce/MapReduceOperation.java
index 3464228..39864b3 100644
--- a/src/main/java/MapReduce/MapReduceOperation.java
+++ b/src/main/java/MapReduce/MapReduceOperation.java
@@ -256,7 +256,7 @@ final public class MapReduceOperation extends MapReducePlan {
job.setGroupingComparatorClass(MRContainerKeyComparator.class);
job.setOutputFormatClass(SequenceFileOutputFormat.class);
for (DataSource p: source.source)
- MultipleInputs.addInputPath(job,new Path(p.path),p.inputFormat,MRMapper.class);
+ 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))
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/MapReduce/MapReduceParsedInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReduceParsedInputFormat.java b/src/main/java/MapReduce/MapReduceParsedInputFormat.java
new file mode 100644
index 0000000..e588dbd
--- /dev/null
+++ b/src/main/java/MapReduce/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();
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/MapReduce/MapReducePlan.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReducePlan.java b/src/main/java/MapReduce/MapReducePlan.java
index bd024b2..ae243cd 100644
--- a/src/main/java/MapReduce/MapReducePlan.java
+++ b/src/main/java/MapReduce/MapReducePlan.java
@@ -52,7 +52,7 @@ public class MapReducePlan extends Plan {
Function accumulator = functional_argument(Plan.conf,acc_fnc);
Tuple pair = new Tuple(2);
for ( DataSource s: S.source )
- if (s.inputFormat != BinaryInputFormat.class) {
+ if (s.inputFormat != MapReduceBinaryInputFormat.class) {
pair.set(0,res);
pair.set(1,aggregate(acc_fnc,zero,
MapOperation.cMap(Interpreter.identity_mapper,acc_fnc,zero,
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/MapReduce/ParsedInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/ParsedInputFormat.java b/src/main/java/MapReduce/ParsedInputFormat.java
deleted file mode 100644
index 6085e53..0000000
--- a/src/main/java/MapReduce/ParsedInputFormat.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/**
- * 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 ParsedInputFormat extends MRQLFileInputFormat {
-
- 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
- */
- 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();
- }
- });
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Bag.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Bag.java b/src/main/java/core/Bag.java
index 77426e4..d0d77c3 100644
--- a/src/main/java/core/Bag.java
+++ b/src/main/java/core/Bag.java
@@ -236,7 +236,7 @@ public class Bag extends MRData implements Iterable<MRData> {
if (!spilled() && Config.hadoop_mode)
try {
if (Plan.conf == null)
- Plan.conf = Evaluator.new_configuration();
+ Plan.conf = Evaluator.evaluator.new_configuration();
final FileSystem fs = FileSystem.getLocal(Plan.conf);
path = new_path(fs);
System.err.println("*** Spilling a Bag to a local file: "+path);
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/BinaryDataSource.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/BinaryDataSource.java b/src/main/java/core/BinaryDataSource.java
index 34c6414..d4338ec 100644
--- a/src/main/java/core/BinaryDataSource.java
+++ b/src/main/java/core/BinaryDataSource.java
@@ -23,11 +23,11 @@ import org.apache.hadoop.conf.Configuration;
/** A DataSource used for storing intermediate results and data dumps */
final public class BinaryDataSource extends DataSource {
BinaryDataSource ( int source_num, String path, Configuration conf ) {
- super(source_num,path,BinaryInputFormat.class,conf);
+ super(source_num,path,Evaluator.evaluator.binaryInputFormat(),conf);
}
BinaryDataSource ( String path, Configuration conf ) {
- super(-1,path,BinaryInputFormat.class,conf);
+ super(-1,path,Evaluator.evaluator.binaryInputFormat(),conf);
}
public String toString () {
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Config.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Config.java b/src/main/java/core/Config.java
index a801320..a957d89 100644
--- a/src/main/java/core/Config.java
+++ b/src/main/java/core/Config.java
@@ -169,6 +169,12 @@ final public class Config {
throw new Error("Expected number of reductions");
nodes = Integer.parseInt(args[i]);
i++;
+ } else if (args[i].equals("-bsp")) {
+ bsp_mode = true;
+ i++;
+ } else if (args[i].equals("-spark")) {
+ spark_mode = true;
+ i++;
} else if (args[i].equals("-bsp_tasks")) {
if (++i >= args.length && Integer.parseInt(args[i]) < 1)
throw new Error("Expected max number of bsp tasks > 1");
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Evaluator.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Evaluator.java b/src/main/java/core/Evaluator.java
new file mode 100644
index 0000000..bc7488e
--- /dev/null
+++ b/src/main/java/core/Evaluator.java
@@ -0,0 +1,94 @@
+/**
+ * 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.io.*;
+import org.apache.hadoop.conf.Configuration;
+
+
+/** Evaluates physical plans using one of the evaluation engines */
+abstract public class Evaluator extends Interpreter {
+
+ /** the current MRQL evaluator */
+ public static Evaluator evaluator;
+
+ /** initialize the evaluator */
+ abstract public void init ( Configuration conf );
+
+ /** shutdown the evaluator */
+ abstract public void shutdown ( Configuration conf );
+
+ /** initialize the query evaluation */
+ abstract public void initialize_query ();
+
+ /** create a new evaluation configuration */
+ abstract public Configuration new_configuration ();
+
+ /** synchronize peers in BSP mode */
+ public MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
+ throw new Error("You can only synchronize BSP tasks");
+ }
+
+ /** distribute a bag among peers in BSP mode */
+ public Bag distribute ( MR_string peerName, Bag s ) {
+ throw new Error("You can only distribute bags among BSP tasks");
+ }
+
+ /** run a BSP task */
+ public MRData bsp ( Tree plan, Environment env ) throws Exception {
+ 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 ();
+
+ /** return the FileInputFormat for binary files */
+ abstract public Class<? extends MRQLFileInputFormat> binaryInputFormat ();
+
+ /** return the FileInputFormat for data generator files */
+ abstract public Class<? extends MRQLFileInputFormat> generatorInputFormat ();
+
+ /** 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
+ */
+ abstract public MRData aggregate ( Tree acc_fnc,
+ Tree zero,
+ Tree plan,
+ Environment env ) throws Exception;
+
+ /** Evaluate a loop a fixed number of times */
+ abstract public Tuple loop ( Tree e, Environment env ) throws Exception;
+
+ /** Evaluate a MRQL physical plan and print tracing info
+ * @param e the physical plan
+ * @param env contains bindings fro variables to values (MRData)
+ * @return a DataSet (stored in HDFS)
+ */
+ abstract public DataSet eval ( final Tree e,
+ final Environment env,
+ final String counter );
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/GeneratorDataSource.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/GeneratorDataSource.java b/src/main/java/core/GeneratorDataSource.java
index 8d93399..380f50d 100644
--- a/src/main/java/core/GeneratorDataSource.java
+++ b/src/main/java/core/GeneratorDataSource.java
@@ -25,11 +25,11 @@ import org.apache.hadoop.conf.Configuration;
/** A DataSource used for processing the range min..max */
final public class GeneratorDataSource extends DataSource {
GeneratorDataSource ( int source_num, String path, Configuration conf ) {
- super(source_num,path,GeneratorInputFormat.class,conf);
+ super(source_num,path,Evaluator.evaluator.generatorInputFormat(),conf);
}
GeneratorDataSource ( String path, Configuration conf ) {
- super(-1,path,GeneratorInputFormat.class,conf);
+ super(-1,path,Evaluator.evaluator.generatorInputFormat(),conf);
}
public static long size ( Path path, Configuration conf ) throws IOException {
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Interpreter.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/Interpreter.gen b/src/main/java/core/Interpreter.gen
index 5cc50ce..a93db7d 100644
--- a/src/main/java/core/Interpreter.gen
+++ b/src/main/java/core/Interpreter.gen
@@ -241,7 +241,7 @@ public class Interpreter extends TypeInference {
Function loop_fnc = new Function () {
public MRData eval ( MRData s ) {
new_global_binding(nm,s);
- MRData val = new MR_dataset(Evaluator.eval(body,new Environment(nm,s,env),nm));
+ MRData val = new MR_dataset(Evaluator.evaluator.eval(body,new Environment(nm,s,env),nm));
return val;
}; };
return MapReduceAlgebra.repeat(loop_fnc,(Bag)evalE(s,env),((MR_int)evalE(n,env)).get());
@@ -303,17 +303,17 @@ public class Interpreter extends TypeInference {
case Collect(`s):
try {
if (Config.hadoop_mode)
- return Plan.collect(Evaluator.eval(s,env,"-"));
+ return Plan.collect(Evaluator.evaluator.eval(s,env,"-"));
Bag b = evalS(s,env);
b.materialize();
return b;
} catch (Exception ex) { throw new Error(ex); }
case dataset_size(`x):
- return new MR_long(Plan.size(Evaluator.eval(x,env,"-")) / (1024*1024));
+ return new MR_long(Plan.size(Evaluator.evaluator.eval(x,env,"-")) / (1024*1024));
case synchronize(`peer,`b):
- return Evaluator.synchronize(((MR_string)evalE(peer,env)),(MR_bool)evalE(b,env));
+ return Evaluator.evaluator.synchronize(((MR_string)evalE(peer,env)),(MR_bool)evalE(b,env));
case distribute(`peer,`s):
- return Evaluator.distribute(((MR_string)evalE(peer,env)),(Bag)evalE(s,env));
+ return Evaluator.evaluator.distribute(((MR_string)evalE(peer,env)),(Bag)evalE(s,env));
case mapReduce(`m,`r,`s,_):
return MapReduceAlgebra.mapReduce(evalF(m,env),
evalF(r,env),
@@ -378,11 +378,11 @@ public class Interpreter extends TypeInference {
(Bag)evalE(s,env));
case Aggregate(`acc,`zero,`s):
if (Config.hadoop_mode)
- return Evaluator.aggregate(closure(acc,env),zero,s,env);
+ return Evaluator.evaluator.aggregate(closure(acc,env),zero,s,env);
else return MapReduceAlgebra.aggregate(evalF(acc,env),evalE(zero,env),evalM(s,env));
case BSP(tuple(...ns),`superstep,`state,`o,...as):
if (Config.hadoop_mode)
- return Evaluator.bsp(e,env);
+ return Evaluator.evaluator.bsp(e,env);
Bag[] ds = new Bag[as.length()];
for ( int i = 0; i < ds.length; i++ )
ds[i] = evalM(as.nth(i),env);
@@ -396,7 +396,7 @@ public class Interpreter extends TypeInference {
ds);
case BSP(`n,`superstep,`state,`o,...as):
if (Config.hadoop_mode)
- return Evaluator.bsp(e,env);
+ return Evaluator.evaluator.bsp(e,env);
Bag[] ds = new Bag[as.length()];
for ( int i = 0; i < ds.length; i++ )
ds[i] = evalM(as.nth(i),env);
@@ -407,7 +407,7 @@ public class Interpreter extends TypeInference {
ds);
case Loop(lambda(tuple(...vs),tuple(...bs)),tuple(...ss),`num):
if (Config.hadoop_mode)
- return Evaluator.loop(e,env);
+ return Evaluator.evaluator.loop(e,env);
int limit = ((MR_int)evalE(num,env)).get();
Bag[] s = new Bag[vs.length()];
for ( int i = 0; i < vs.length(); i++ )
@@ -466,7 +466,7 @@ public class Interpreter extends TypeInference {
case _:
try {
if (Config.hadoop_mode)
- return new MR_dataset(Evaluator.eval(e,env,"-"));
+ return new MR_dataset(Evaluator.evaluator.eval(e,env,"-"));
else return evalS(e,env);
} catch (Exception ex) { throw new Error(ex); }
};
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/LineParser.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/LineParser.gen b/src/main/java/core/LineParser.gen
index f4af146..fb1e068 100644
--- a/src/main/java/core/LineParser.gen
+++ b/src/main/java/core/LineParser.gen
@@ -116,7 +116,7 @@ final public class LineParser implements Parser {
public void initialize ( Trees args ) {
if (Config.hadoop_mode && Plan.conf == null)
- Plan.conf = Evaluator.new_configuration();
+ Plan.conf = Evaluator.evaluator.new_configuration();
if (args.length() != 2)
throw new Error("The line parser must have two arguments: "+args);
if (!(args.nth(0) instanceof StringLeaf))
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/MRQLFileInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MRQLFileInputFormat.java b/src/main/java/core/MRQLFileInputFormat.java
new file mode 100644
index 0000000..6b6aec2
--- /dev/null
+++ b/src/main/java/core/MRQLFileInputFormat.java
@@ -0,0 +1,36 @@
+/**
+ * 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.*;
+
+
+/** An interface for all MRQL FileInputFormats */
+public interface MRQLFileInputFormat {
+ /** materialize the input file into a memory Bag */
+ 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 Bag collect ( final DataSet x, boolean strip ) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Main.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Main.java b/src/main/java/core/Main.java
index 978222a..e3f7eba 100644
--- a/src/main/java/core/Main.java
+++ b/src/main/java/core/Main.java
@@ -58,10 +58,20 @@ final public class Main {
public static void main ( String[] args ) throws Exception {
boolean hadoop = false;
- for ( String arg: args )
+ for ( String arg: args ) {
hadoop |= arg.equals("-local") || arg.equals("-dist");
+ Config.bsp_mode |= arg.equals("-bsp");
+ Config.spark_mode |= arg.equals("-spark");
+ };
+ Config.map_reduce_mode = !Config.bsp_mode && !Config.spark_mode;
+ if (Config.map_reduce_mode)
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.MapReduceEvaluator").newInstance();
+ if (Config.bsp_mode)
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.BSPEvaluator").newInstance();
+ if (Config.spark_mode)
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.SparkEvaluator").newInstance();
if (hadoop) {
- conf = Evaluator.new_configuration();
+ conf = Evaluator.evaluator.new_configuration();
GenericOptionsParser gop = new GenericOptionsParser(conf,args);
conf = gop.getConfiguration();
args = gop.getRemainingArgs();
@@ -73,7 +83,7 @@ final public class Main {
((Logger)en.nextElement()).setLevel(Level.WARN);
LogManager.getRootLogger().setLevel(Level.WARN);
};
- Evaluator.init(conf);
+ Evaluator.evaluator.init(conf);
new TopLevel();
System.out.print("Apache MRQL version "+version+" (");
if (Config.compile_functional_arguments)
@@ -134,7 +144,7 @@ final public class Main {
} finally {
if (hadoop) {
Plan.clean();
- Evaluator.shutdown(Plan.conf);
+ Evaluator.evaluator.shutdown(Plan.conf);
};
if (Config.compile_functional_arguments)
Compiler.clean();
@@ -154,7 +164,7 @@ final public class Main {
} finally {
if (hadoop) {
Plan.clean();
- Evaluator.shutdown(Plan.conf);
+ Evaluator.evaluator.shutdown(Plan.conf);
};
if (Config.compile_functional_arguments)
Compiler.clean();
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/MapReduceAlgebra.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MapReduceAlgebra.java b/src/main/java/core/MapReduceAlgebra.java
index bc2990e..9b50ada 100644
--- a/src/main/java/core/MapReduceAlgebra.java
+++ b/src/main/java/core/MapReduceAlgebra.java
@@ -214,6 +214,7 @@ final public class MapReduceAlgebra {
*/
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);
@@ -226,9 +227,10 @@ final public class MapReduceAlgebra {
for ( MRData y: Y ) {
MRData key = ky.eval(y);
Bag match = hashTable.get(key);
- if (match != null)
- for ( MRData x: match )
- res.add(f.eval(new Tuple(x,y)));
+ 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)));
};
return res;
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/ParsedDataSource.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/ParsedDataSource.java b/src/main/java/core/ParsedDataSource.java
index 7ac4c16..e000405 100644
--- a/src/main/java/core/ParsedDataSource.java
+++ b/src/main/java/core/ParsedDataSource.java
@@ -33,7 +33,7 @@ public class ParsedDataSource extends DataSource {
Class<? extends Parser> parser,
Trees args,
Configuration conf ) {
- super(source_num,path,ParsedInputFormat.class,conf);
+ super(source_num,path,Evaluator.evaluator.parsedInputFormat(),conf);
this.parser = parser;
this.args = args;
}
@@ -42,7 +42,7 @@ public class ParsedDataSource extends DataSource {
Class<? extends Parser> parser,
Trees args,
Configuration conf ) {
- super(-1,path,ParsedInputFormat.class,conf);
+ super(-1,path,Evaluator.evaluator.parsedInputFormat(),conf);
this.parser = parser;
this.args = args;
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Plan.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Plan.java b/src/main/java/core/Plan.java
index f5fb869..6ecdafa 100644
--- a/src/main/java/core/Plan.java
+++ b/src/main/java/core/Plan.java
@@ -298,12 +298,12 @@ public class Plan {
/** The collect physical operator */
public final static Bag collect ( final DataSet x, boolean strip ) throws Exception {
- return MRQLFileInputFormat.collect(x,strip);
+ return Evaluator.evaluator.parsedInputFormat().newInstance().collect(x,strip);
}
/** The collect physical operator */
public final static Bag collect ( final DataSet x ) throws Exception {
- return MRQLFileInputFormat.collect(x,true);
+ return collect(x,true);
}
/** the DataSet union physical operator */
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/SystemFunctions.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/SystemFunctions.java b/src/main/java/core/SystemFunctions.java
index ab82998..5c0ed4b 100644
--- a/src/main/java/core/SystemFunctions.java
+++ b/src/main/java/core/SystemFunctions.java
@@ -41,11 +41,11 @@ final public class SystemFunctions {
static Random random = new Random();
public static MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
- return Evaluator.synchronize(peerName,mr_exit);
+ return Evaluator.evaluator.synchronize(peerName,mr_exit);
}
public static Bag distribute ( MR_string peerName, Bag s ) {
- return Evaluator.distribute(peerName,s);
+ return Evaluator.evaluator.distribute(peerName,s);
}
public static MR_bool lt ( MR_short x, MR_short y ) { return (x.get() < y.get()) ? true_value : false_value; }
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/Test.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Test.java b/src/main/java/core/Test.java
index 85598fb..78cb0d5 100644
--- a/src/main/java/core/Test.java
+++ b/src/main/java/core/Test.java
@@ -86,18 +86,28 @@ final public class Test {
public static void main ( String[] args ) throws Exception {
boolean hadoop = false;
- for ( String arg: args )
+ for ( String arg: args ) {
hadoop |= arg.equals("-local") || arg.equals("-dist");
+ Config.bsp_mode |= arg.equals("-bsp");
+ Config.spark_mode |= arg.equals("-spark");
+ };
+ Config.map_reduce_mode = !Config.bsp_mode && !Config.spark_mode;
+ if (Config.map_reduce_mode)
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.MapReduceEvaluator").newInstance();
+ if (Config.bsp_mode)
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.BSPEvaluator").newInstance();
+ if (Config.spark_mode)
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.SparkEvaluator").newInstance();
Config.quiet_execution = true;
if (hadoop) {
- conf = Evaluator.new_configuration();
+ conf = Evaluator.evaluator.new_configuration();
GenericOptionsParser gop = new GenericOptionsParser(conf,args);
conf = gop.getConfiguration();
args = gop.getRemainingArgs();
};
Config.parse_args(args,conf);
Config.hadoop_mode = Config.local_mode || Config.distributed_mode;
- Evaluator.init(conf);
+ Evaluator.evaluator.init(conf);
new TopLevel();
Config.testing = true;
if (Config.hadoop_mode && Config.bsp_mode)
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/core/TopLevel.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/TopLevel.gen b/src/main/java/core/TopLevel.gen
index 5c86f37..9972b86 100644
--- a/src/main/java/core/TopLevel.gen
+++ b/src/main/java/core/TopLevel.gen
@@ -62,7 +62,7 @@ final public class TopLevel extends Interpreter {
if (plan == null)
return null;
if (Config.hadoop_mode)
- Evaluator.initialize_query();
+ Evaluator.evaluator.initialize_query();
MRData res = evalE(plan,null);
if (print) {
if (!Config.quiet_execution)
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/17a0ed95/src/main/java/spark/BinaryInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/spark/BinaryInputFormat.java b/src/main/java/spark/BinaryInputFormat.java
deleted file mode 100644
index ad364bb..0000000
--- a/src/main/java/spark/BinaryInputFormat.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/**
- * 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.mapred.*;
-
-
-/** Input format for hadoop sequence files */
-final public class BinaryInputFormat extends MRQLFileInputFormat {
- public static class BinaryInputRecordReader extends SequenceFileRecordReader<MRContainer,MRContainer> {
- final MRContainer result = new MRContainer();
-
- public BinaryInputRecordReader ( FileSplit split,
- JobConf job ) throws IOException {
- super(job,split);
- }
-
- @Override
- public synchronized boolean next ( MRContainer key, MRContainer value ) throws IOException {
- boolean b = super.next(key,result);
- value.set(result.data());
- return b;
- }
- }
-
- @Override
- public RecordReader<MRContainer,MRContainer>
- getRecordReader ( InputSplit split,
- JobConf job,
- Reporter reporter ) throws IOException {
- String path = ((FileSplit)split).getPath().toString();
- BinaryDataSource ds = (BinaryDataSource)DataSource.get(path,job);
- return new BinaryInputRecordReader((FileSplit)split,job);
- }
-}