| /** |
| * 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("mapreduce.framework.name","local"); |
| conf.set("fs.defaultFS","file:///"); // doesn't work for hadoop 1.* |
| } else { |
| if (!System.getenv("MAPRED_JOB_TRACKER").equals("")) |
| conf.set("mapred.job.tracker",System.getenv("MAPRED_JOB_TRACKER")); |
| if (!System.getenv("FS_DEFAULT_NAME").equals("")) |
| conf.set("fs.defaultFS",System.getenv("FS_DEFAULT_NAME")); // doesn't work for hadoop 1.* |
| } |
| } |
| |
| /** 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(); |
| } |
| |
| final public static Configuration clear_configuration ( Configuration old_conf ) { |
| Configuration conf = new Configuration(); |
| evaluator.init(conf); |
| // keep the current mrql configurations |
| for ( Map.Entry<String,String> entry: old_conf ) |
| if (entry.getKey().startsWith("mrql.") && entry.getValue() != null) |
| conf.set(entry.getKey(),entry.getValue()); |
| // old Configuration doesn't have unset |
| // conf.unset("mapred.input.dir.formats"); // need to reset MultipleInputs for local mode |
| return conf; |
| } |
| |
| /** 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; |
| } |
| |
| /** Coerce a persistent collection to a Bag */ |
| public Bag toBag ( MRData data ) { |
| try { |
| if (data instanceof MR_dataset) |
| return Plan.collect(((MR_dataset)data).dataset()); |
| return (Bag)data; |
| } catch (Exception ex) { |
| throw new Error("Cannot coerce "+data+" to a Bag: "+ex); |
| } |
| } |
| |
| /** 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(Config.nodes)); |
| int m = n; |
| // 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 trace(`msg,`tp,`x): |
| long n = pre_trace(((MR_string)evalE(msg,env)).get()); |
| DataSet ds = evalD(x,env,counter); |
| if (!counter.equals("-")) // if inside a repeat loop |
| match tp { |
| case `T(tuple(`etp,bool)): |
| tp = #<`T(`etp)>; |
| }; |
| trace(n,tp,new MR_dataset(ds)); |
| return ds; |
| 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 nth(`u,`n): |
| MRData x = ((Tuple)evalE(u,env)).get((int)n.longValue()); |
| if (x instanceof MR_dataset) |
| return ((MR_dataset)x).dataset(); |
| throw new Error("Evaluation error in: "+print_query(e)); |
| 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)); |
| } |
| } |
| } |
