Spork hadoop asm jars incompatibility fix

bin/pig

@@ -52,12 +52,15 @@ case "`uname`" in
 CYGWIN*) cygwin=true;;
 esac
 debug=false
+SPORK=false
 
 remaining=()
 includeHCatalog="";
 addJarString=-Dpig.additional.jars\=;
 additionalJars="";
 # filter command line parameter
+
+
 for f in "$@"; do
      if [[ $f == "-secretDebugCmd" || $f == "-printCmdDebug" ]]; then
         debug=true
@@ -73,6 +76,10 @@ for f in "$@"; do
       else
         remaining[${#remaining[@]}]="$f"
      fi
+     ####for spark mode#####
+     if [[ ${remaining[@]} == "-x spark" ]]; then  ####
+      SPORK=true
+     fi
 done
 
 # resolve links - $0 may be a softlink
@@ -340,7 +347,8 @@ if [ "$includeHCatalog" == "true" ]; then
 fi
 
 # run it
-if [ -n "$HADOOP_BIN" ]; then
+#####modified for spark mode
+if [ -n "$HADOOP_BIN" ] && [ "$SPORK" == "false" ]; then 
     if [ "$debug" == "true" ]; then
         echo "Find hadoop at $HADOOP_BIN"
     fi
@@ -356,6 +364,7 @@ if [ -n "$HADOOP_BIN" ]; then
         PIG_JAR=`echo $PIG_HOME/share/pig/pig-*withouthadoop.jar`
     fi
 
+
     if [ -n "$PIG_JAR" ]; then
         CLASSPATH=${CLASSPATH}:$PIG_JAR
     else
@@ -370,7 +379,6 @@ if [ -n "$HADOOP_BIN" ]; then
         echo "HADOOP_CLASSPATH: $HADOOP_CLASSPATH"
         echo "HADOOP_OPTS: $HADOOP_OPTS"
         echo "$HADOOP_BIN" jar "$PIG_JAR" "${remaining[@]}"
-        echo
     else
         exec "$HADOOP_BIN" jar "$PIG_JAR" "${remaining[@]}"
     fi

ivy.xml

@@ -278,10 +278,11 @@
     <dependency org="org.mockito" name="mockito-all" rev="${mockito.version}" conf="compile->default"/>
 
     <!-- <dependency org="org.apache.spark" name="spark-core_2.9.3" rev="0.8.2-incubating-SNAPSHOT-TW1" conf="compile->default">  -->
-    <dependency org="org.apache.spark" name="spark-core_2.9.3" rev="0.8.1-incubating" conf="compile->default">
+    <!-- <dependency org="org.apache.spark" name="spark-core_2.9.3" rev="0.8.1-incubating" conf="compile->default"> -->
+   <dependency org="org.apache.spark" name="spark-core_2.10" rev="0.9.0-incubating" conf="compile->default">
       <exclude org="org.eclipse.jetty.orbit" module="javax.servlet"/>
     </dependency>
-    <dependency org="org.scala-lang" name="scala-library" rev="2.9.3" conf="compile"/>
+    <dependency org="org.scala-lang" name="scala-library" rev="2.10.3" conf="compile"/>
 
     </dependencies>
 </ivy-module>

pig-spark

@@ -18,12 +18,16 @@ export SPARK_YARN_APP_JAR=pig/pig-withouthadoop.jar
 export SPARK_JAR=spark.jar
 export SPARK_MASTER=yarn-client
 
+export SPARK_HOME=
+
 # jars to ship, pig-withouthadoop.jar to workaround Classloader issue
 export SPARK_JARS=piggybank.jar
 
 # Pig settings
 export PIG_CLASSPATH=${SPARK_JAR}:${SPARK_JARS}:mesos.jar:pig-withouthadoop.jar
 export SPARK_PIG_JAR=pig/pig-withouthadoop.jar
+export PIG_CLASSPATH=${SPARK_JAR}:${SPARK_JARS}:mesos.jar:pig-withouthadoop.jar
+export SPARK_PIG_JAR=pig-withouthadoop.jar
 
 # Cluster settings
 export SPARK_WORKER_CORES=4

src/org/apache/pig/backend/hadoop/executionengine/spark/SparkUtil.java

@@ -13,8 +13,13 @@
 import scala.Product2;
 import scala.collection.JavaConversions;
 import scala.collection.Seq;
-import scala.reflect.ClassManifest;
-import scala.reflect.ClassManifest$;
+
+//==============FFFFFF=================
+//import scala.reflect.ClassManifest;
+//import scala.reflect.ClassManifest$;
+import scala.reflect.ClassTag;
+import scala.reflect.ClassTag$;
+
 import org.apache.spark.rdd.RDD;
 
 import java.io.IOException;
@@ -25,18 +30,18 @@
  */
 public class SparkUtil {
 
-    public static <T> ClassManifest<T> getManifest(Class<T> clazz) {
-        return ClassManifest$.MODULE$.fromClass(clazz);
+    public static <T> ClassTag<T> getManifest(Class<T> clazz) {
+        return ClassTag$.MODULE$.apply(clazz);
     }
 
     @SuppressWarnings("unchecked")
-    public static <K,V> ClassManifest<Tuple2<K, V>> getTuple2Manifest() {
-        return (ClassManifest<Tuple2<K, V>>)(Object)getManifest(Tuple2.class);
+    public static <K,V> ClassTag<Tuple2<K, V>> getTuple2Manifest() {
+        return (ClassTag<Tuple2<K, V>>)(Object)getManifest(Tuple2.class);
     }
 
     @SuppressWarnings("unchecked")
-    public static <K,V> ClassManifest<Product2<K, V>> getProduct2Manifest() {
-        return (ClassManifest<Product2<K, V>>)(Object)getManifest(Product2.class);
+    public static <K,V> ClassTag<Product2<K, V>> getProduct2Manifest() {
+        return (ClassTag<Product2<K, V>>)(Object)getManifest(Product2.class);
     }
 
     public static JobConf newJobConf(PigContext pigContext) throws IOException {

src/org/apache/pig/backend/hadoop/executionengine/spark/converter/DistinctConverter.java

@@ -13,7 +13,10 @@
 import scala.Function1;
 import scala.Function2;
 import scala.Tuple2;
-import scala.reflect.ClassManifest;
+
+//import scala.reflect.ClassManifest;
+import scala.reflect.ClassTag;
+
 import scala.runtime.AbstractFunction1;
 import scala.runtime.AbstractFunction2;
 import org.apache.spark.rdd.PairRDDFunctions;
@@ -33,7 +36,7 @@ public RDD<Tuple> convert(List<RDD<Tuple>> predecessors, PODistinct poDistinct)
         SparkUtil.assertPredecessorSize(predecessors, poDistinct, 1);
         RDD<Tuple> rdd = predecessors.get(0);
 
-        ClassManifest<Tuple2<Tuple, Object>> tuple2ClassManifest = SparkUtil.<Tuple, Object>getTuple2Manifest();
+        ClassTag<Tuple2<Tuple, Object>> tuple2ClassManifest = SparkUtil.<Tuple, Object>getTuple2Manifest();
 
         RDD<Tuple2<Tuple, Object>> rddPairs = rdd.map(TO_KEY_VALUE_FUNCTION, tuple2ClassManifest);
         PairRDDFunctions<Tuple, Object> pairRDDFunctions =

src/org/apache/pig/backend/hadoop/executionengine/spark/converter/GlobalRearrangeConverter.java

@@ -22,7 +22,9 @@
 import scala.Tuple2;
 import scala.collection.JavaConversions;
 import scala.collection.Seq;
-import scala.reflect.ClassManifest;
+
+//import scala.reflect.ClassManifest;
+import scala.reflect.ClassTag;
 
 @SuppressWarnings({ "serial"})
 public class GlobalRearrangeConverter implements POConverter<Tuple, Tuple, POGlobalRearrange> {
@@ -59,7 +61,7 @@ public RDD<Tuple> convert(List<RDD<Tuple>> predecessors,
         } else {
             //COGROUP
             // each pred returns (index, key, value)
-            ClassManifest<Tuple2<Object, Tuple>> tuple2ClassManifest = SparkUtil.<Object, Tuple>getTuple2Manifest();
+            ClassTag<Tuple2<Object, Tuple>> tuple2ClassManifest = SparkUtil.<Object, Tuple>getTuple2Manifest();
 
             List<RDD<Tuple2<Object, Tuple>>> rddPairs = new ArrayList();
             for (RDD<Tuple> rdd : predecessors) {