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/10/27 21:38:15 UTC
git commit: MRQL-56: Improve total aggregations and repetitions with
shared results in Flink mode
Repository: incubator-mrql
Updated Branches:
refs/heads/master 192e58c30 -> fe2079e82
MRQL-56: Improve total aggregations and repetitions with shared results in Flink mode
Project: http://git-wip-us.apache.org/repos/asf/incubator-mrql/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-mrql/commit/fe2079e8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-mrql/tree/fe2079e8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-mrql/diff/fe2079e8
Branch: refs/heads/master
Commit: fe2079e8281e1d955bee4e7bcb2fa2852ab7533e
Parents: 192e58c
Author: fegaras <fe...@cse.uta.edu>
Authored: Mon Oct 27 15:37:48 2014 -0500
Committer: fegaras <fe...@cse.uta.edu>
Committed: Mon Oct 27 15:37:48 2014 -0500
----------------------------------------------------------------------
conf/mrql-env.sh | 9 +-
.../main/java/org/apache/mrql/Interpreter.gen | 18 ++
.../java/org/apache/mrql/FlinkEvaluator.gen | 169 +++++++++++++++++--
pom.xml | 2 +-
queries/pi.mrql | 24 +++
src/main/assembly/bin.xml | 3 +
src/main/assembly/src.xml | 3 +
7 files changed, 213 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/conf/mrql-env.sh
----------------------------------------------------------------------
diff --git a/conf/mrql-env.sh b/conf/mrql-env.sh
index 902d910..afd367e 100644
--- a/conf/mrql-env.sh
+++ b/conf/mrql-env.sh
@@ -58,10 +58,11 @@ FS_DEFAULT_NAME=hdfs://localhost:9000/
# Optional: Hama configuration. Supports versions 0.6.2, 0.6.3, and 0.6.4
HAMA_VERSION=0.6.4
# The Hadoop installation directory
-HAMA_HOME=${HOME}/hama-${HAMA_VERSION}
+#HAMA_HOME=${HOME}/hama-${HAMA_VERSION}
+HAMA_HOME=${HOME}/hama-${HAMA_VERSION}-yarn/dist/target/hama-0.6.4/hama-0.6.4/
# The Hama configuration (as defined in hama-site.xml)
-BSP_MASTER_ADDRESS=`hostname`:40000
-HAMA_ZOOKEEPER_QUORUM=`hostname`
+BSP_MASTER_ADDRESS=localhost:40000
+HAMA_ZOOKEEPER_QUORUM=localhost
# Optional: Spark configuration. Supports versions 1.0.0, 1.0.2, and 1.1.0
@@ -86,7 +87,7 @@ SPARK_MASTER_MEMORY=512M
# Optional: Flink configuration. Supports versions 0.6-incubating, 0.6.1-incubating, and 0.7.0-incubating
# Note: for yarn, set yarn.nodemanager.vmem-check-enabled to false in yarn-site.xml
-FLINK_VERSION=yarn-0.6.1-incubating
+FLINK_VERSION=yarn-0.7.0-incubating
# Flink installation directory
FLINK_HOME=${HOME}/flink-${FLINK_VERSION}
# (use this for a Flink snapshot):
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/core/src/main/java/org/apache/mrql/Interpreter.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Interpreter.gen b/core/src/main/java/org/apache/mrql/Interpreter.gen
index b0f0ff3..86a36f7 100644
--- a/core/src/main/java/org/apache/mrql/Interpreter.gen
+++ b/core/src/main/java/org/apache/mrql/Interpreter.gen
@@ -135,6 +135,19 @@ public class Interpreter extends TypeInference {
throw new Error("wrong MRData: "+x);
}
+ private static boolean has_source ( Tree e ) {
+ match e {
+ case Generator(...): return true;
+ case BinarySource(...): return true;
+ case ParsedSource(...): return true;
+ case `f(...as):
+ for ( Tree a: as )
+ if (has_source(a))
+ return true;
+ };
+ return false;
+ }
+
/** evaluate an MRQL function in memory */
private final static Function evalf ( final String v,
final Tree body,
@@ -286,6 +299,11 @@ public class Interpreter extends TypeInference {
return MapReduceAlgebra.filter(evalF(p,env),evalF(m,env),(Bag)evalE(s,env));
case map(`m,`s):
return MapReduceAlgebra.map(evalF(m,env),(Bag)evalE(s,env));
+ case repeat(...):
+ if (!Config.flink_mode || !Config.hadoop_mode || !has_source(e))
+ fail;
+ // send Flink repetitions to the Flink evaluator (they need Flink iterations)
+ return Evaluator.evaluator.aggregate(#<0>,#<0>,e,env);
case repeat(lambda(`v,`b),`s,`n):
final String nm = v.toString();
final Tree body = b;
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen b/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
index d14def5..1ccba9e 100644
--- a/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
+++ b/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
@@ -19,6 +19,7 @@ package org.apache.mrql;
import java.util.List;
import java.util.ArrayList;
+import java.util.Collection;
import java.io.*;
import java.net.URI;
import java.net.URL;
@@ -39,7 +40,7 @@ import org.apache.flink.api.java.operators.*;
/** Evaluates physical plans in Apache Flink mode */
-final public class FlinkEvaluator extends Evaluator {
+final public class FlinkEvaluator extends Evaluator implements Serializable {
final static URL flink_jar = FlinkEvaluator.class.getProtectionDomain().getCodeSource().getLocation();
public static ExecutionEnvironment flink_env;
// an HDFS tmp file used to hold the data source directory information in distributed mode
@@ -187,6 +188,39 @@ final public class FlinkEvaluator extends Evaluator {
}
}
+ /* broadcast the loop value to all nodes */
+ public static final class set_environment extends RichMapFunction<FData,FData> {
+ String var;
+
+ set_environment ( String v ) { var = v; }
+
+ @Override
+ public void open ( org.apache.flink.configuration.Configuration parameters ) throws Exception {
+ final Bag value = bag(getRuntimeContext().getBroadcastVariable(var));
+ Interpreter.new_global_binding(var.toString(),value);
+ }
+
+ @Override
+ public FData map ( FData value ) throws Exception {
+ return new FData(((Tuple)value.data()).first());
+ }
+ }
+
+ private static Bag bag ( Collection<Object> values ) {
+ Bag s = new Bag();
+ for ( Object value: values )
+ s.add(((FData)value).data());
+ return s;
+ }
+
+ /* convert a Bag into a DataSet */
+ private static DataSet<FData> dataset ( Bag s ) {
+ ArrayList<FData> a = new ArrayList(s.size());
+ for ( MRData x: s )
+ a.add(new FData(x));
+ return flink_env.fromCollection(a);
+ }
+
/** The Aggregate physical operator
* @param merge the accumulator function from (T,T) to T
* @param zero the zero element of type T
@@ -198,23 +232,61 @@ final public class FlinkEvaluator extends Evaluator {
Tree zero,
Tree plan,
Environment env ) throws Exception {
- MRData z = evalE(zero,env);
+ final MRData z = evalE(zero,env);
Tree type = TypeInference.type_inference(zero);
match plan {
+ case repeat(lambda(`v,`b),`s,`n):
+ // Special case for Flink: repetition where the loop result must be shared among nodes
+ final int max_num = ((MR_int)evalE(n,null)).get();
+ final Bag init = (Bag)evalE(s,env);
+ Interpreter.new_global_binding(v.toString(),init);
+ final DataSet<FData> ds = dataset(init);
+ final IterativeDataSet<FData> startOfIteration = ds.iterate(max_num);
+ final DataSet<FData> step = eval(b,env);
+ final DataSet<FData> toBeFedBack = step.map(new set_environment(v.toString()))
+ .withBroadcastSet(startOfIteration,v.toString());
+ // Flink doesn't allow termination criteria for kmeans-like iterations:
+ // (java.lang.IllegalStateException: Error: Iterative task without a single iterative input)
+ //final DataSet<FData> terminationCriterion = step.filter(new repeat_filter());
+ final DataSet<FData> res = startOfIteration.closeWith(toBeFedBack);//,terminationCriterion);
+ String path = absolute_path(Plan.new_path(Plan.conf));
+ res.write(new FlinkOutputFormat(),path);
+ flink_env.execute("MRQL repeat");
+ Bag bs = new Bag();
+ for ( MRData x: new FlinkDataSource(res,path,false).take(-1) )
+ bs.add(x);
+ return bs;
case AggregateMap(`m,`acc,_,`s):
- return aggregate(evalD(#<cMap(`m,`s)>,env),z,acc,merge,type);
+ final DataSet<FData> ds = eval(s,env).mapPartition(new aggregate_partition(m,acc,z));
+ return aggregate(ds,z,merge,merge,type);
case MapAggregateReduce(`m,`r,`acc,_,`s,`o):
if (acc.equals(#<null>))
fail;
- return aggregate(evalD(#<MapReduce(`m,`r,`s,`o)>,env),z,acc,merge,type);
+ final DataSet<FData> ds = eval(s,env).flatMap(cmap_fnc(m));
+ final DataSet<FData> gs = ds.groupBy(new KeySelector<FData,FData>() {
+ @Override
+ public FData getKey ( FData value ) {
+ return new FData(((Tuple)value.data()).first());
+ }
+ }).reduceGroup(new groupBy_combiner_reducer(r,acc,z));
+ return aggregate(gs,z,merge,merge,type);
case CrossAggregateProduct(`mx,`my,`r,`acc,_,`x,`y):
if (acc.equals(#<null>))
fail;
- return aggregate(evalD(#<CrossProduct(`mx,`my,`r,`x,`y)>,env),z,acc,merge,type);
+ final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
+ final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
+ final DataSet<FData> ds = xs.cross(ys).with(new cross_combiner_reducer(r,acc,z));
+ return aggregate(ds,z,merge,merge,type);
case MapAggregateReduce2(`mx,`my,`r,`acc,_,`x,`y,`o):
if (acc.equals(#<null>))
fail;
- return aggregate(evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env),z,acc,merge,type);
+ final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
+ final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
+ final DataSet<FData> ds = xs.coGroup(ys)
+ .where(new join_key())
+ .equalTo(new join_key())
+ .with(new join_combiner_reducer(r,acc,z));
+ return aggregate(ds,z,merge,merge,type);
case MapAggregateJoin(`mx,`my,`r,`acc,_,`x,`y):
if (acc.equals(#<null>))
fail;
@@ -294,9 +366,8 @@ final public class FlinkEvaluator extends Evaluator {
final DataSet<FData> res = evalD(e,env);
if (Config.trace_execution)
try {
- List<FData> out = new ArrayList<FData>();
- res.output(new LocalCollectionOutputFormat(out));
- System.out.println(tabs(tab_count)+"-> "+out);
+ System.out.println(tabs(tab_count)+"->");
+ res.first(Config.max_bag_size_print).print();
tab_count -= 3;
} catch (Exception ex) {
System.err.println(ex.getMessage());
@@ -318,7 +389,7 @@ final public class FlinkEvaluator extends Evaluator {
}
/* convert the stream of pairs to a bag of their second element */
- public static Bag bag ( Iterable<FData> values ) {
+ private static Bag bag ( Iterable<FData> values ) {
Bag s = new Bag();
for ( FData value: values )
s.add(((Tuple)value.data()).second());
@@ -415,6 +486,47 @@ final public class FlinkEvaluator extends Evaluator {
}
}
+ public static final class aggregate_partition extends RichMapPartitionFunction<FData,FData> {
+ final Function m;
+ final Function acc;
+ final MRData zero;
+
+ aggregate_partition ( Tree m, Tree acc, MRData zero ) {
+ this.m = evalF(m,null);
+ this.acc = evalF(acc,null);
+ this.zero = zero;
+ }
+
+ @Override
+ public void mapPartition ( Iterable<FData> values, Collector<FData> out ) throws Exception {
+ MRData aggregated_value = zero;
+ for ( FData value: values )
+ for ( MRData v: (Bag)m.eval(value.data()) )
+ aggregated_value = acc.eval(new Tuple(aggregated_value,v));
+ out.collect(new FData(aggregated_value));
+ }
+ }
+
+ public static final class groupBy_combiner_reducer extends RichGroupReduceFunction<FData,FData> {
+ final Function r;
+ final Function acc;
+ final MRData zero;
+
+ groupBy_combiner_reducer ( Tree r, Tree acc, MRData zero ) {
+ this.r = evalF(r,null);
+ this.acc = evalF(acc,null);
+ this.zero = zero;
+ }
+
+ @Override
+ public void reduce ( final Iterable<FData> values, Collector<FData> out ) {
+ MRData value = zero;
+ for ( MRData v: (Bag)r.eval(new Tuple(bag(values))) )
+ value = acc.eval(new Tuple(value,v));
+ out.collect(new FData(value));
+ }
+ }
+
public static final class join_reducer extends RichCoGroupFunction<FData,FData,FData> {
final Function f;
@@ -429,6 +541,26 @@ final public class FlinkEvaluator extends Evaluator {
}
}
+ public static final class join_combiner_reducer extends RichCoGroupFunction<FData,FData,FData> {
+ final Function r;
+ final Function acc;
+ final MRData zero;
+
+ join_combiner_reducer ( Tree r, Tree acc, MRData zero ) {
+ this.r = evalF(r,null);
+ this.acc = evalF(acc,null);
+ this.zero = zero;
+ }
+
+ @Override
+ public void coGroup ( Iterable<FData> xs, Iterable<FData> ys, Collector<FData> out ) {
+ MRData value = zero;
+ for ( MRData v: (Bag)r.eval(new Tuple(bag(xs),bag(ys))) )
+ value = acc.eval(new Tuple(value,v));
+ out.collect(new FData(value));
+ }
+ }
+
public static final class cross_reducer extends RichCrossFunction<FData,FData,FData> {
final Function f;
@@ -442,6 +574,23 @@ final public class FlinkEvaluator extends Evaluator {
}
}
+ public static final class cross_combiner_reducer extends RichCrossFunction<FData,FData,FData> {
+ final Function r;
+ final Function acc;
+ final MRData zero;
+
+ cross_combiner_reducer ( Tree r, Tree acc, MRData zero ) {
+ this.r = evalF(r,null);
+ this.acc = evalF(acc,null);
+ this.zero = zero;
+ }
+
+ @Override
+ public FData cross ( FData x, FData y ) throws Exception {
+ return new FData(acc.eval(new Tuple(zero,r.eval(new Tuple(x.data(),y.data())))));
+ }
+ }
+
public static final class mapjoin_reducer extends RichJoinFunction<FData,FData,FData> {
final Function f;
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index a0dc930..10995c8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -60,7 +60,7 @@
<hama.version>0.6.4</hama.version>
<spark.version>1.1.0</spark.version>
<scala.version>2.10</scala.version>
- <flink.version>0.6.1-incubating</flink.version>
+ <flink.version>0.7.0-incubating</flink.version>
<skipTests>true</skipTests>
</properties>
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/queries/pi.mrql
----------------------------------------------------------------------
diff --git a/queries/pi.mrql b/queries/pi.mrql
new file mode 100644
index 0000000..0b781d1
--- /dev/null
+++ b/queries/pi.mrql
@@ -0,0 +1,24 @@
+/**
+ * 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.
+ */
+
+points = toLong(args[0]);
+
+count(select i
+ from i in 1 ... points
+ where pow(random(10000)/10000.0,2) + pow(random(10000)/10000.0,2) < 1.0
+ )*(4.0 as double)/points;
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/src/main/assembly/bin.xml
----------------------------------------------------------------------
diff --git a/src/main/assembly/bin.xml b/src/main/assembly/bin.xml
index 612d0ca..b9b927e 100644
--- a/src/main/assembly/bin.xml
+++ b/src/main/assembly/bin.xml
@@ -59,6 +59,9 @@
<directory>../spark/src</directory>
</fileSet>
<fileSet>
+ <directory>../flink/src</directory>
+ </fileSet>
+ <fileSet>
<directory>../queries</directory>
</fileSet>
<fileSet>
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/fe2079e8/src/main/assembly/src.xml
----------------------------------------------------------------------
diff --git a/src/main/assembly/src.xml b/src/main/assembly/src.xml
index 18f37c5..faac771 100644
--- a/src/main/assembly/src.xml
+++ b/src/main/assembly/src.xml
@@ -65,6 +65,9 @@
<directory>../spark/src</directory>
</fileSet>
<fileSet>
+ <directory>../flink/src</directory>
+ </fileSet>
+ <fileSet>
<directory>../queries</directory>
</fileSet>
<fileSet>