bsp/src/main/java/org/apache/mrql/BSPEvaluator.gen - incubator-retired-mrql - Git at Google

 /**
  * 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.util.List;
 import java.io.*;
 import org.apache.hadoop.io.*;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hama.HamaConfiguration;


 /** Evaluate a physical plan in BSP mode using Apache Hama */
 final public class BSPEvaluator extends Evaluator {

     /** initialize the BSP evaluator */
     final public void init ( Configuration conf ) {
         Config.bsp_mode = true;
         if (Config.hadoop_mode)
             if (Config.local_mode) {
                 conf.set("bsp.master.address","local");
                 conf.set("hama.zookeeper.quorum","localhost");
                 conf.setInt("bsp.local.tasks.maximum",Config.nodes);
                 conf.set("fs.defaultFS","file:///");
             } else {
                 conf.set("bsp.master.address",System.getenv("BSP_MASTER_ADDRESS"));
                 conf.set("hama.zookeeper.quorum",System.getenv("HAMA_ZOOKEEPER_QUORUM"));
                 conf.setInt("bsp.local.tasks.maximum",Config.nodes);
                 if (!System.getenv("FS_DEFAULT_NAME").equals(""))
                     conf.set("fs.defaultFS",System.getenv("FS_DEFAULT_NAME"));
             }
     }

     final public void initialize_query () {
     }

     /** shutdown the BSP evaluator */
     final public void shutdown ( Configuration conf ) {
     }

     final public Configuration new_configuration () {
         return new HamaConfiguration();
     }

     final public MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
         if (Config.hadoop_mode)
             return BSPPlan.BSPop.synchronize(peerName,mr_exit);
         else return mr_exit;
     }

     final public Bag distribute ( MR_string peerName, Bag s ) {
         if (Config.hadoop_mode)
             return BSPPlan.BSPop.distribute(peerName,s);
         else return s;
     }

     /** Evaluate MRQL physical operators using BSP
      * @param plan the physical plan
      * @param env contains bindings fro variables to values (MRData)
      * @return a DataSet (stored in HDFS)
      */
     final public MRData bsp ( Tree plan, Environment env ) throws Exception {
         match plan {
         case BSP(tuple(...ns),`superstep,`init_state,`o,...S):
             int[] os = new int[ns.length()];
             for ( int i = 0; i < os.length; i++ )
                 os[i] = (int)((LongLeaf)ns.nth(i)).value();
             DataSet ds = eval(S.head(),env);
             for ( Tree s: S.tail() )
                 ds.merge(eval(s,env));
             return BSPPlan.BSP(os,
                                closure(superstep,env),
                                init_state,
                                o.equals(#<true>),
                                ds);
         case BSP(`n,`superstep,`init_state,`o,...S):
             DataSet ds = eval(S.head(),env);
             for ( Tree s: S.tail() )
                 ds.merge(eval(s,env));
             return BSPPlan.BSP(new int[]{(int)((LongLeaf)n).value()},
                                closure(superstep,env),
                                init_state,
                                o.equals(#<true>),
                                ds);
         }
         throw new Error("Cannot perform the BSP plan: "+plan);
     }

     /** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
     final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
         return BSPParsedInputFormat.class;
     }

     /** return the FileInputFormat for binary files */
     final public Class<? extends MRQLFileInputFormat> binaryInputFormat () {
         return BSPBinaryInputFormat.class;
     }

     /** return the FileInputFormat for data generator files */
     final public Class<? extends MRQLFileInputFormat> generatorInputFormat () {
         return BSPGeneratorInputFormat.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());
         } catch (Exception ex) {
             throw new Error("Cannot coerce "+data+" to a Bag");
         };
         return (Bag)data;
     }

     /** 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 {
         match plan {
         case BSP(`n,`superstep,`init_state,`o,...S):
             DataSet ds = eval(S.head(),env,"-");
             for ( Tree s: S.tail() )
                 ds.merge(eval(s,env,"-"));
             return BSPPlan.BSPaggregate((int)((LongLeaf)n).value(),
                                         closure(superstep,env),
                                         init_state,
                                         closure(acc_fnc,env),
                                         zero,
                                         ds);
         }
         throw new Error("Cannot perform the aggregation: "+plan);
     }

     /** 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");
     }

     final public DataSet eval ( final Tree e,
                                 final Environment env,
                                 final String counter ) {
         return eval(e,env);
     }

     /** Evaluate MRQL BSP physical operators using Hama (returns a DataSet)
      *   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 ) {
         if (Config.trace_execution) {
             tab_count += 3;
             System.out.println(tabs(tab_count)+print_query(e));
         };
         DataSet res = evalD(e,env);
         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 BSP physical operators using Hama (returns a DataSet)
      * @param e the physical plan
      * @param env contains bindings fro variables to values (MRData)
      * @return a DataSet (stored in HDFS)
      */
     final public DataSet evalD ( final Tree e, final Environment env ) {
         try {
             match e {
             case BSPSource(`n,BinarySource(`file,_)):
                 return Plan.binarySource((int)((LongLeaf)n).value(),file.stringValue());
             case BSPSource(`n,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((int)((LongLeaf)n).value(),
                                          p,((MR_string)evalE(file,env)).get(),args);
             case BSPSource(`n,Generator(`min,`max,`size)):
                 return Plan.generator((int)((LongLeaf)n).value(),
                                       ((MR_long)evalE(min,env)).get(),
                                       ((MR_long)evalE(max,env)).get(),
                                       ((MR_long)evalE(size,env)).get());
             case BSPSource(`n,`v):
                 if (!v.is_variable())
                     fail;
                 DataSet ds = evalD(v,env);
                 for ( DataSource s: ds.source )
                     s.source_num = (int)((LongLeaf)n).value();
                 return ds;
             case Merge(`x,`y):
                 return Plan.merge(eval(x,env),eval(y,env));
             case Dump(`s):
                 return Plan.fileCache((Bag)evalE(s,env));
             case trace(`msg,`tp,`x):
                 long n = pre_trace(((MR_string)evalE(msg,env)).get());
                 DataSet ds = evalD(x,env);
                 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");
             };
             MRData d = bsp(e,env);
             if (d instanceof MR_dataset)
                 return ((MR_dataset)d).dataset();
             throw new Error("Cannot evaluate the BSP 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));
         }
     }
 }
	/**
	* 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.util.List;
	import java.io.*;
	import org.apache.hadoop.io.*;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hama.HamaConfiguration;


	/** Evaluate a physical plan in BSP mode using Apache Hama */
	final public class BSPEvaluator extends Evaluator {

	/** initialize the BSP evaluator */
	final public void init ( Configuration conf ) {
	Config.bsp_mode = true;
	if (Config.hadoop_mode)
	if (Config.local_mode) {
	conf.set("bsp.master.address","local");
	conf.set("hama.zookeeper.quorum","localhost");
	conf.setInt("bsp.local.tasks.maximum",Config.nodes);
	conf.set("fs.defaultFS","file:///");
	} else {
	conf.set("bsp.master.address",System.getenv("BSP_MASTER_ADDRESS"));
	conf.set("hama.zookeeper.quorum",System.getenv("HAMA_ZOOKEEPER_QUORUM"));
	conf.setInt("bsp.local.tasks.maximum",Config.nodes);
	if (!System.getenv("FS_DEFAULT_NAME").equals(""))
	conf.set("fs.defaultFS",System.getenv("FS_DEFAULT_NAME"));
	}
	}

	final public void initialize_query () {
	}

	/** shutdown the BSP evaluator */
	final public void shutdown ( Configuration conf ) {
	}

	final public Configuration new_configuration () {
	return new HamaConfiguration();
	}

	final public MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
	if (Config.hadoop_mode)
	return BSPPlan.BSPop.synchronize(peerName,mr_exit);
	else return mr_exit;
	}

	final public Bag distribute ( MR_string peerName, Bag s ) {
	if (Config.hadoop_mode)
	return BSPPlan.BSPop.distribute(peerName,s);
	else return s;
	}

	/** Evaluate MRQL physical operators using BSP
	* @param plan the physical plan
	* @param env contains bindings fro variables to values (MRData)
	* @return a DataSet (stored in HDFS)
	*/
	final public MRData bsp ( Tree plan, Environment env ) throws Exception {
	match plan {
	case BSP(tuple(...ns),`superstep,`init_state,`o,...S):
	int[] os = new int[ns.length()];
	for ( int i = 0; i < os.length; i++ )
	os[i] = (int)((LongLeaf)ns.nth(i)).value();
	DataSet ds = eval(S.head(),env);
	for ( Tree s: S.tail() )
	ds.merge(eval(s,env));
	return BSPPlan.BSP(os,
	closure(superstep,env),
	init_state,
	o.equals(#<true>),
	ds);
	case BSP(`n,`superstep,`init_state,`o,...S):
	DataSet ds = eval(S.head(),env);
	for ( Tree s: S.tail() )
	ds.merge(eval(s,env));
	return BSPPlan.BSP(new int[]{(int)((LongLeaf)n).value()},
	closure(superstep,env),
	init_state,
	o.equals(#<true>),
	ds);
	}
	throw new Error("Cannot perform the BSP plan: "+plan);
	}

	/** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
	final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
	return BSPParsedInputFormat.class;
	}

	/** return the FileInputFormat for binary files */
	final public Class<? extends MRQLFileInputFormat> binaryInputFormat () {
	return BSPBinaryInputFormat.class;
	}

	/** return the FileInputFormat for data generator files */
	final public Class<? extends MRQLFileInputFormat> generatorInputFormat () {
	return BSPGeneratorInputFormat.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());
	} catch (Exception ex) {
	throw new Error("Cannot coerce "+data+" to a Bag");
	};
	return (Bag)data;
	}

	/** 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 {
	match plan {
	case BSP(`n,`superstep,`init_state,`o,...S):
	DataSet ds = eval(S.head(),env,"-");
	for ( Tree s: S.tail() )
	ds.merge(eval(s,env,"-"));
	return BSPPlan.BSPaggregate((int)((LongLeaf)n).value(),
	closure(superstep,env),
	init_state,
	closure(acc_fnc,env),
	zero,
	ds);
	}
	throw new Error("Cannot perform the aggregation: "+plan);
	}

	/** 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");
	}

	final public DataSet eval ( final Tree e,
	final Environment env,
	final String counter ) {
	return eval(e,env);
	}

	/** Evaluate MRQL BSP physical operators using Hama (returns a DataSet)
	* 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 ) {
	if (Config.trace_execution) {
	tab_count += 3;
	System.out.println(tabs(tab_count)+print_query(e));
	};
	DataSet res = evalD(e,env);
	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 BSP physical operators using Hama (returns a DataSet)
	* @param e the physical plan
	* @param env contains bindings fro variables to values (MRData)
	* @return a DataSet (stored in HDFS)
	*/
	final public DataSet evalD ( final Tree e, final Environment env ) {
	try {
	match e {
	case BSPSource(`n,BinarySource(`file,_)):
	return Plan.binarySource((int)((LongLeaf)n).value(),file.stringValue());
	case BSPSource(`n,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((int)((LongLeaf)n).value(),
	p,((MR_string)evalE(file,env)).get(),args);
	case BSPSource(`n,Generator(`min,`max,`size)):
	return Plan.generator((int)((LongLeaf)n).value(),
	((MR_long)evalE(min,env)).get(),
	((MR_long)evalE(max,env)).get(),
	((MR_long)evalE(size,env)).get());
	case BSPSource(`n,`v):
	if (!v.is_variable())
	fail;
	DataSet ds = evalD(v,env);
	for ( DataSource s: ds.source )
	s.source_num = (int)((LongLeaf)n).value();
	return ds;
	case Merge(`x,`y):
	return Plan.merge(eval(x,env),eval(y,env));
	case Dump(`s):
	return Plan.fileCache((Bag)evalE(s,env));
	case trace(`msg,`tp,`x):
	long n = pre_trace(((MR_string)evalE(msg,env)).get());
	DataSet ds = evalD(x,env);
	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");
	};
	MRData d = bsp(e,env);
	if (d instanceof MR_dataset)
	return ((MR_dataset)d).dataset();
	throw new Error("Cannot evaluate the BSP 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));
	}
	}
	}