diff --git a/bin/mrql.spark b/bin/mrql.spark
index a422e18..239182a 100755
--- a/bin/mrql.spark
+++ b/bin/mrql.spark
@@ -28,7 +28,7 @@
 
 . "$MRQL_HOME/conf/mrql-env.sh"
 
-if [[ !(-f ${SPARK_JARS}) ]]; then
+if [[ -z ${SPARK_JARS} ]]; then
    echo "*** Cannot find the Spark jar file. Need to edit mrql-env.sh"; exit -1
 fi
 
diff --git a/conf/mrql-env.sh b/conf/mrql-env.sh
index ed891e1..aefd589 100644
--- a/conf/mrql-env.sh
+++ b/conf/mrql-env.sh
@@ -78,10 +78,10 @@
 BSP_SPLIT_INPUT=
 
 
-# Optional: Spark configuration. Supports versions 1.0.0, 1.0.2, 1.1.0, 1.1.1, 1.2.0, 1.3.0, 1.3.1, 1.6.0, and 1.6.2
-# (Spark versions 0.8.1, 0.9.0, and 0.9.1 are supported by MRQL 0.9.0)
+# Optional: Spark configuration. Supports versions 1.*, and 2.0.0
+# For Spark 2.*, use: mvn -Dspark2
+# Spark versions 0.8.1, 0.9.0, and 0.9.1 are supported by MRQL 0.9.0 only.
 # You may use the Spark prebuilts bin-hadoop1 or bin-hadoop2 (Yarn)
-# For distributed mode, give write permission to /tmp: hadoop fs -chmod -R 777 /tmp
 # Tested in local, standalone deploy, and Yarn modes
 SPARK_HOME=${HOME}/spark-1.6.2-bin-hadoop2.6
 # URI of the Spark master node:
@@ -123,10 +123,13 @@
    SPARK_JARS=`ls ${SPARK_HOME}/assembly/target/scala-*/*.jar`
 else if [[ -d ${SPARK_HOME}/lib ]]; then
    SPARK_JARS=`ls ${SPARK_HOME}/lib/spark-assembly-*.jar`
+else if [[ -d ${SPARK_HOME}/jars ]]; then
+   SPARK_JARS=`ls ${SPARK_HOME}/jars/spark-core*.jar`
+fi
 fi
 fi
 
-HADOOP_JARS=`$HADOOP_HOME/bin/hadoop classpath`
+HADOOP_JARS=`${HADOOP_HOME}/bin/hadoop classpath`
 
 if [[ !(-f ${CUP_JAR}) ]]; then
    echo "*** Cannot find the parser generator CUP jar file. Need to edit mrql-env.sh"; exit -1
diff --git a/core/src/main/java/org/apache/mrql/Compiler.gen b/core/src/main/java/org/apache/mrql/Compiler.gen
index 5dbaf24..51f4534 100644
--- a/core/src/main/java/org/apache/mrql/Compiler.gen
+++ b/core/src/main/java/org/apache/mrql/Compiler.gen
@@ -164,7 +164,7 @@
             compiled_lambdas = #[ ];
             user_functions_num = lambda_num++;
             // remove the old jar
-            if (jar_path != null)
+            if (false && jar_path != null) // Spark 2.* needs the old jar
                 remove(new File(jar_path));
             jar_path = tmp_dir+"/mrql_args_"+(new Random().nextInt(1000000))+".jar";
             out = new StringBuffer(1000);
diff --git a/spark/pom.xml b/spark/pom.xml
index 614e707..e53aef9 100644
--- a/spark/pom.xml
+++ b/spark/pom.xml
@@ -152,4 +152,69 @@
       </plugin>
     </plugins>
   </build>
+
+  <profiles>
+    <profile>
+      <id>spark1</id>
+      <activation>
+        <property>
+          <name>!spark2</name>
+        </property>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>build-helper-maven-plugin</artifactId>
+            <version>1.5</version>
+            <executions>
+              <execution>
+                <id>add-sources</id>
+                <phase>generate-sources</phase>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>spark1</source>
+                  </sources>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+    <profile>
+      <id>spark2</id>
+      <activation>
+        <property>
+          <name>spark2</name>
+        </property>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>build-helper-maven-plugin</artifactId>
+            <version>1.5</version>
+            <executions>
+              <execution>
+                <id>add-sources</id>
+                <phase>generate-sources</phase>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>spark2</source>
+                  </sources>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
 </project>
diff --git a/spark/spark1/Spark1.java b/spark/spark1/Spark1.java
new file mode 100644
index 0000000..efa719b
--- /dev/null
+++ b/spark/spark1/Spark1.java
@@ -0,0 +1,56 @@
+/**
+ * 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.util.Iterator;
+import scala.Tuple2;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+
+
+abstract class FmFunction<T,R> implements FlatMapFunction<T,R> {
+    abstract Iterator<R> eval ( T t ) throws Exception;
+
+    public Iterable<R> call ( final T t ) throws Exception {
+        return new Iterable<R>() {
+            public Iterator<R> iterator() {
+                try {
+                    return eval(t);
+                } catch (Exception ex) {
+                    throw new Error(ex);
+                }
+            }
+        };
+    }
+}
+
+abstract class PairFmFunction<T,K,V> implements PairFlatMapFunction<T,K,V> {
+    abstract Iterator<Tuple2<K,V>> eval ( T t ) throws Exception;
+
+    public Iterable<Tuple2<K,V>> call ( final T t ) throws Exception {
+        return new Iterable<Tuple2<K,V>>() {
+            public Iterator<Tuple2<K,V>> iterator() {
+                try {
+                    return eval(t);
+                } catch (Exception ex) {
+                    throw new Error(ex);
+                }
+            }
+        };
+    }
+}
diff --git a/spark/spark2/Spark2.java b/spark/spark2/Spark2.java
new file mode 100644
index 0000000..82793bd
--- /dev/null
+++ b/spark/spark2/Spark2.java
@@ -0,0 +1,40 @@
+/**
+ * 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.util.Iterator;
+import scala.Tuple2;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+
+
+abstract class FmFunction<T,R> implements FlatMapFunction<T,R> {
+    abstract Iterator<R> eval ( T t ) throws Exception;
+
+    public Iterator<R> call ( final T t ) throws Exception {
+        return eval(t);
+    }
+}
+
+abstract class PairFmFunction<T,K,V> implements PairFlatMapFunction<T,K,V> {
+    abstract Iterator<Tuple2<K,V>> eval ( T t ) throws Exception;
+
+    public Iterator<Tuple2<K,V>> call ( final T t ) throws Exception {
+        return eval(t);
+    }
+}
diff --git a/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen b/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
index 3ded60e..83de421 100644
--- a/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
+++ b/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
@@ -41,8 +41,6 @@
 import org.apache.spark.api.java.function.Function2;
 import org.apache.spark.api.java.function.Function3;
 import org.apache.spark.api.java.function.PairFunction;
-import org.apache.spark.api.java.function.FlatMapFunction;
-import org.apache.spark.api.java.function.PairFlatMapFunction;
 import org.apache.spark.api.java.function.Function;
 import org.apache.spark.api.java.function.VoidFunction;
 import org.apache.spark.streaming.api.java.JavaStreamingContext;
@@ -388,13 +386,13 @@
     }
 
     /* convert an MRQL lambda to a Spark Function */
-    private static FlatMapFunction<MRData,MRData> cmap_fnc ( Tree fnc, Environment env ) {
+    private static FmFunction<MRData,MRData> cmap_fnc ( Tree fnc, Environment env ) {
         final Broadcast<Environment> broadcast_env = spark_context.broadcast(global_env);
         final org.apache.mrql.Function f = evalF(fnc,env);
-        return new FlatMapFunction<MRData,MRData>() {
-            public Iterable<MRData> call ( MRData value ) {
+        return new FmFunction<MRData,MRData>() {
+            public Iterator<MRData> eval ( MRData value ) {
                 global_env = broadcast_env.value();
-                return (Bag)f.eval(value);
+                return ((Bag)f.eval(value)).iterator();
             }
         };
     }
@@ -443,26 +441,22 @@
             });
     }
 
-    private static Iterable<Tuple2<MRData,MRData>> joinIterator ( final Iterator<MRData> i ) {
-        return new Iterable<Tuple2<MRData,MRData>>() {
-            public Iterator<Tuple2<MRData,MRData>> iterator() {
-                return new Iterator<Tuple2<MRData,MRData>> () {
-                    public Tuple2<MRData,MRData> next () {
-                        Tuple data = (Tuple)i.next();
-                        return new Tuple2<MRData,MRData>(data.first(),data.second());
-                    }
-                    public boolean hasNext () {
-                        return i.hasNext();
-                    }
-                    public void remove () {}
-                };
+    private static Iterator<Tuple2<MRData,MRData>> joinIterator ( final Iterator<MRData> i ) {
+        return new Iterator<Tuple2<MRData,MRData>> () {
+            public Tuple2<MRData,MRData> next () {
+                Tuple data = (Tuple)i.next();
+                return new Tuple2<MRData,MRData>(data.first(),data.second());
             }
+            public boolean hasNext () {
+                return i.hasNext();
+            }
+            public void remove () {}
         };
     }
 
-    private static FlatMapFunction<Iterator<MRData>,MRData> combiner_fnc ( final org.apache.mrql.Function f ) {
-        return new FlatMapFunction<Iterator<MRData>,MRData>() {
-                  public Iterable<MRData> call ( final Iterator<MRData> i ) {
+    private static FmFunction<Iterator<MRData>,MRData> combiner_fnc ( final org.apache.mrql.Function f ) {
+        return new FmFunction<Iterator<MRData>,MRData>() {
+                  public Iterator<MRData> eval ( final Iterator<MRData> i ) {
                       return MapReduceAlgebra.cmap(new org.apache.mrql.Function() {
                               public MRData eval ( MRData x ) {
                                   final MRData key = ((Tuple)x).first();
@@ -483,7 +477,7 @@
                                   public boolean hasNext () {
                                       return i.hasNext();
                                   }
-                              })));
+                              }))).iterator();
                   }
         };
     }
@@ -519,9 +513,9 @@
                 return eval(x,env,rdd_env)
                     .flatMap(cmap_fnc(mx,env))
                     .cartesian(eval(y,env,rdd_env).flatMap(cmap_fnc(my,env)))
-                    .flatMap(new FlatMapFunction<Tuple2<MRData,MRData>,MRData>() {
-                            public Iterable<MRData> call ( Tuple2<MRData,MRData> value ) {
-                                return (Bag)fr.eval(new Tuple(value._1,value._2));
+                    .flatMap(new FmFunction<Tuple2<MRData,MRData>,MRData>() {
+                            public Iterator<MRData> eval ( Tuple2<MRData,MRData> value ) {
+                                return ((Bag)fr.eval(new Tuple(value._1,value._2))).iterator();
                             }
                         });
             case MapReduce2(`mx,`my,`r,`x,`y,`o):
@@ -544,8 +538,8 @@
                                 return new Tuple2(t.first(),t.second());
                             }
                         })
-                    : eval(x,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
+                    : eval(x,env,rdd_env).flatMapToPair(new PairFmFunction<MRData,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( MRData value ) {
                                 return joinIterator(((Bag)fx.eval(value)).iterator());
                             }
                         });
@@ -567,15 +561,15 @@
                                 return new Tuple2(t.first(),t.second());
                             }
                         })
-                    : eval(y,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
+                    : eval(y,env,rdd_env).flatMapToPair(new PairFmFunction<MRData,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( MRData value ) {
                                 return joinIterator(((Bag)fy.eval(value)).iterator());
                             }
                         });
 		return xs.cogroup(ys)
-                    .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
-                            public Iterable<MRData> call ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
-                                return (Bag)fr.eval(new Tuple(bag(value._2._1),bag(value._2._2)));
+                    .flatMap(new FmFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
+                            public Iterator<MRData> eval ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
+                                return ((Bag)fr.eval(new Tuple(bag(value._2._1),bag(value._2._2)))).iterator();
                             }
                         });
             case GroupByJoin(`kx,`ky,`gx,`gy,`acc,`zero,`r,`x,`y,`o):
@@ -592,50 +586,42 @@
                 final MRData one = new MR_byte(1);
                 final MRData two = new MR_byte(2);
                 final JavaPairRDD<MRData,MRData> xs
-                    = eval(x,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( final MRData value ) {
-                                return new Iterable<Tuple2<MRData,MRData>>() {
-                                    public Iterator<Tuple2<MRData,MRData>> iterator() {
-                                        return new Iterator<Tuple2<MRData,MRData>>() {
-                                            int i = 0;
-                                            public Tuple2<MRData,MRData> next () {
-                                                MRData key = new MR_int((fgx.eval(value).hashCode() % m)+m*i);
-                                                i++;
-                                                return new Tuple2<MRData,MRData>(key,new Tuple(one,value));
-                                            }
-                                            public boolean hasNext () {
-                                                return i < n;
-                                            }
-                                            public void remove () {}
-                                        };
+                    = eval(x,env,rdd_env).flatMapToPair(new PairFmFunction<MRData,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( final MRData value ) {
+                                return new Iterator<Tuple2<MRData,MRData>>() {
+                                    int i = 0;
+                                    public Tuple2<MRData,MRData> next () {
+                                        MRData key = new MR_int((fgx.eval(value).hashCode() % m)+m*i);
+                                        i++;
+                                        return new Tuple2<MRData,MRData>(key,new Tuple(one,value));
                                     }
+                                    public boolean hasNext () {
+                                        return i < n;
+                                    }
+                                    public void remove () {}
                                 };
                             }
                         });
                 final JavaPairRDD<MRData,MRData> ys
-                    = eval(y,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( final MRData value ) {
-                                return new Iterable<Tuple2<MRData,MRData>>() {
-                                    public Iterator<Tuple2<MRData,MRData>> iterator() {
-                                        return new Iterator<Tuple2<MRData,MRData>>() {
-                                            int j = 0;
-                                            public Tuple2<MRData,MRData> next () {
-                                                MRData key = new MR_int((fgy.eval(value).hashCode() % n)*m+j);
-                                                j++;
-                                                return new Tuple2<MRData,MRData>(key,new Tuple(two,value));
-                                            }
-                                            public boolean hasNext () {
-                                                return j < m;
-                                            }
-                                            public void remove () {}
-                                        };
+                    = eval(y,env,rdd_env).flatMapToPair(new PairFmFunction<MRData,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( final MRData value ) {
+                                return new Iterator<Tuple2<MRData,MRData>>() {
+                                    int j = 0;
+                                    public Tuple2<MRData,MRData> next () {
+                                        MRData key = new MR_int((fgy.eval(value).hashCode() % n)*m+j);
+                                        j++;
+                                        return new Tuple2<MRData,MRData>(key,new Tuple(two,value));
                                     }
+                                    public boolean hasNext () {
+                                        return j < m;
+                                    }
+                                    public void remove () {}
                                 };
                             }
                         });
                 return xs.union(ys).groupByKey()
-                    .mapPartitions(new FlatMapFunction<Iterator<Tuple2<MRData,Iterable<MRData>>>,MRData>() {
-                        public Iterable<MRData> call ( final Iterator<Tuple2<MRData,Iterable<MRData>>> value ) {
+                    .mapPartitions(new FmFunction<Iterator<Tuple2<MRData,Iterable<MRData>>>,MRData>() {
+                        public Iterator<MRData> eval ( final Iterator<Tuple2<MRData,Iterable<MRData>>> value ) {
                             Bag xb = new Bag();
                             Bag yb = new Bag();
                             while (value.hasNext()) {
@@ -649,18 +635,14 @@
                             };
                             final Bag b = MapReduceAlgebra.mergeGroupByJoin(fkx,fky,fgx,fgy,fc,z,fr,xb,yb);
                             xb = null; yb = null;
-                            return new Iterable<MRData>() {
-                                public Iterator<MRData> iterator() {
-                                    return b.iterator();
-                                }
-                            };
+                            return b.iterator();
                         }
                         });                
             case OuterMerge(`merge,`state,`data):
                 final org.apache.mrql.Function fm = evalF(merge,env);
                 JavaPairRDD<MRData,MRData> S = eval(state,env,rdd_env)
-                    .mapPartitionsToPair(new PairFlatMapFunction<Iterator<MRData>,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( final Iterator<MRData> i ) {
+                    .mapPartitionsToPair(new PairFmFunction<Iterator<MRData>,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( final Iterator<MRData> i ) {
                                 return joinIterator(i);
                             }
                         },true);   // do not repartition the state S
@@ -672,8 +654,8 @@
                             }
                         });
                 JavaRDD<MRData> res = S.cogroup(ds)
-                    .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
-                            public Iterable<MRData> call ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
+                    .flatMap(new FmFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
+                            public Iterator<MRData> eval ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
                                 final Iterator<MRData> ix = value._2._1.iterator();
                                 final Iterator<MRData> iy = value._2._2.iterator();
                                 ArrayList<MRData> a = new ArrayList<MRData>();
@@ -683,7 +665,7 @@
                                     else a.add(new Tuple(value._1,ix.next()));
                                 else if (iy.hasNext())
                                     a.add(new Tuple(value._1,iy.next()));
-                                return a;
+                                return a.iterator();
                             }
                         }).cache();
                 cached_rdds.add(res);
@@ -691,8 +673,8 @@
             case RightOuterMerge(`merge,`state,`data):
                 final org.apache.mrql.Function fm = evalF(merge,env);
                 JavaPairRDD<MRData,MRData> S = eval(state,env,rdd_env)
-                    .mapPartitionsToPair(new PairFlatMapFunction<Iterator<MRData>,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( final Iterator<MRData> i ) {
+                    .mapPartitionsToPair(new PairFmFunction<Iterator<MRData>,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( final Iterator<MRData> i ) {
                                 return joinIterator(i);
                             }
                         },true);   // do not repartition the state S
@@ -704,17 +686,17 @@
                             }
                         });
                 JavaRDD<MRData> res = S.cogroup(ds)
-                    .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
-                            public Iterable<MRData> call ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
+                    .flatMap(new FmFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
+                            public Iterator<MRData> eval ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
                                 final Iterator<MRData> ix = value._2._1.iterator();
                                 final Iterator<MRData> iy = value._2._2.iterator();
-                                return (ix.hasNext())
+                                return ((ix.hasNext())
                                     ? ((iy.hasNext())
                                        ? new Bag(new Tuple(value._1,fm.eval(new Tuple(ix.next(),iy.next()))))
                                        : new Bag())
                                     : ((iy.hasNext())
                                        ? new Bag(new Tuple(value._1,iy.next()))
-                                       : new Bag());
+                                       : new Bag())).iterator();
                             }
                         }).cache();
                 cached_rdds.add(res);
@@ -731,23 +713,19 @@
                                 return new Tuple2(t.first(),t.second());
                             }
                         }).collectAsMap());
-                return eval(x,env,rdd_env).mapPartitions(new FlatMapFunction<Iterator<MRData>,MRData>() {
-                        public Iterable<MRData> call ( final Iterator<MRData> i ) {
+                return eval(x,env,rdd_env).mapPartitions(new FmFunction<Iterator<MRData>,MRData>() {
+                        public Iterator<MRData> eval ( final Iterator<MRData> i ) {
                             final Map<MRData,MRData> m = ys.value();
-                            return new Iterable<MRData>() {
-                                public Iterator<MRData> iterator() {
-                                    return new Iterator<MRData> () {
-                                        public MRData next () {
-                                            final Tuple p = (Tuple)fx.eval(i.next());
-                                            final MRData pd = m.get(p.first());
-                                            return ((Bag)fr.eval(new Tuple(p.second(),(pd == null) ? empty_bag : pd)));
-                                        }
-                                        public boolean hasNext () {
-                                            return i.hasNext();
-                                        }
-                                        public void remove () {}
-                                    };
+                            return new Iterator<MRData> () {
+                                public MRData next () {
+                                    final Tuple p = (Tuple)fx.eval(i.next());
+                                    final MRData pd = m.get(p.first());
+                                    return ((Bag)fr.eval(new Tuple(p.second(),(pd == null) ? empty_bag : pd)));
                                 }
+                                public boolean hasNext () {
+                                    return i.hasNext();
+                                }
+                                public void remove () {}
                             };
                         }
                     },true);
@@ -757,39 +735,35 @@
                 final org.apache.mrql.Function fr = evalF(r,env);
                 final Broadcast<Map<MRData,MRData>> ys
                     = spark_context.broadcast(eval(y,env,rdd_env)
-                         .flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
-                            public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
+                         .flatMapToPair(new PairFmFunction<MRData,MRData,MRData>() {
+                            public Iterator<Tuple2<MRData,MRData>> eval ( MRData value ) {
                                 return joinIterator(((Bag)fy.eval(value)).iterator());
                             }
                         }).collectAsMap());
-                return eval(x,env,rdd_env).flatMap(new FlatMapFunction<MRData,MRData>() {
-                        public Iterable<MRData> call ( MRData value ) {
+                return eval(x,env,rdd_env).flatMap(new FmFunction<MRData,MRData>() {
+                        public Iterator<MRData> eval ( MRData value ) {
                             final Map<MRData,MRData> m = ys.value();
                             final Iterator<MRData> i = ((Bag)fx.eval(value)).iterator();
-                            return new Iterable<MRData>() {
-                                public Iterator<MRData> iterator() {
-                                    return new Iterator<MRData>() {
-                                        Tuple p;
-                                        Iterator<MRData> ix = null;
-                                        public MRData next () {
-                                            return ix.next();
-                                        }
-                                        public boolean hasNext () {
-                                            if (ix != null && ix.hasNext())
-                                                return true;
-                                            while (i.hasNext()) {
-                                                p = (Tuple)i.next();
-                                                MRData pd = m.get(p.first());
-                                                Bag bb = ((Bag)fr.eval(new Tuple(p.second(),(pd == null) ? empty_bag : pd)));
-                                                ix = bb.iterator();
-                                                if (ix.hasNext())
-                                                    return true;
-                                            };
-                                            return false;
-                                        }
-                                        public void remove () {}
-                                    };
+                            return new Iterator<MRData>() {
+                                Tuple p;
+                                Iterator<MRData> ix = null;
+                                public MRData next () {
+                                    return ix.next();
                                 }
+                                public boolean hasNext () {
+                                    if (ix != null && ix.hasNext())
+                                        return true;
+                                    while (i.hasNext()) {
+                                        p = (Tuple)i.next();
+                                        MRData pd = m.get(p.first());
+                                        Bag bb = ((Bag)fr.eval(new Tuple(p.second(),(pd == null) ? empty_bag : pd)));
+                                        ix = bb.iterator();
+                                        if (ix.hasNext())
+                                            return true;
+                                    };
+                                    return false;
+                                }
+                                public void remove () {}
                             };
                         }
                     });
diff --git a/spark/src/main/java/org/apache/mrql/SparkStreaming.gen b/spark/src/main/java/org/apache/mrql/SparkStreaming.gen
index 1ae28b1..2b74b3e 100644
--- a/spark/src/main/java/org/apache/mrql/SparkStreaming.gen
+++ b/spark/src/main/java/org/apache/mrql/SparkStreaming.gen
@@ -159,7 +159,11 @@
             plan = get_streams(plan,env);
             stream_processing(plan,env,dataset_env,f);
         };
-        stream_context.start();
-        stream_context.awaitTermination();
+        try {
+            stream_context.start();
+            stream_context.awaitTermination();
+        } catch (Exception ex) {
+            throw new Error(ex);
+        }
     }
 }