at2

Author: Elena Akhmatova

spark/src/main/java/com/matthewrathbone/sparktest/SparkJoins.java

@@ -0,0 +1,89 @@
+package main.java.com.matthewrathbone.sparktest;
+
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+
+import org.apache.hadoop.mapred.TextOutputFormat;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.Function2;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+import org.apache.spark.api.java.function.PairFunction;
+import org.apache.spark.network.shuffle.ShuffleClient;
+
+
+import scala.Tuple2;
+import scala.Predef;
+import com.google.common.base.Optional;
+
+public class SparkJoins {
+    @SuppressWarnings("serial")
+    
+    public static final PairFunction<Tuple2<Integer, Optional<String>>, Integer, String> KEY_VALUE_PAIRER =
+    new PairFunction<Tuple2<Integer, Optional<String>>, Integer, String>() {
+    	public Tuple2<Integer, String> call(
+    			Tuple2<Integer, Optional<String>> a) throws Exception {
+			// a._2.isPresent()
+    		return new Tuple2<Integer, String>(a._1, a._2.get());
+    	}
+	};
+	
+	public static JavaRDD<Tuple2<Integer,Optional<String>>> joinData(JavaPairRDD<Integer, Integer> t, JavaPairRDD<Integer, String> u){
+        //Left Outer join operation
+        JavaRDD<Tuple2<Integer,Optional<String>>> leftJoinOutput = t.leftOuterJoin(u).values().distinct();
+        //System.out.println("LeftOuterJoins function Output: "+leftJoinOutput.collect());
+        return leftJoinOutput;
+	}
+	
+	public static JavaPairRDD<Integer, String> modifyData(JavaRDD<Tuple2<Integer,Optional<String>>> d){
+		return d.mapToPair(KEY_VALUE_PAIRER);
+		//System.out.println("MapToPair function Output: "+res.collect());
+	}
+	
+	public static Map<Integer, Object> countData(JavaPairRDD<Integer, String> d){
+        //System.out.println("MapToPair function Output: "+res.collect());
+        Map<Integer, Object> result = d.countByKey();
+        //System.out.println("CountByKey function Output: "+result.toString());
+        return result;
+	}
+	
+    
+    public static void main(String[] args) throws FileNotFoundException {
+    	// SPARK_USER
+        JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("SparkJoins").setMaster("local"));
+
+        JavaRDD<String> transactionInputFile = sc.textFile(args[0]);
+        JavaPairRDD<Integer, Integer> transactionPairs = transactionInputFile.mapToPair(new PairFunction<String, Integer, Integer>() {
+            public Tuple2<Integer, Integer> call(String s) {
+                String[] transactionSplit = s.split("\t");
+                return new Tuple2<Integer, Integer>(Integer.valueOf(transactionSplit[2]), Integer.valueOf(transactionSplit[1]));
+            }
+        });
+        
+        JavaRDD<String> customerInputFile = sc.textFile(args[1]);
+        JavaPairRDD<Integer, String> customerPairs = customerInputFile.mapToPair(new PairFunction<String, Integer, String>() {
+            public Tuple2<Integer, String> call(String s) {
+                String[] customerSplit = s.split("\t");
+                return new Tuple2<Integer, String>(Integer.valueOf(customerSplit[0]), customerSplit[3]);
+            }
+        });
+
+        Map<Integer, Object> result = countData(modifyData(joinData(transactionPairs, customerPairs)));
+        
+        List<Tuple2<Integer, Long>> output = new ArrayList<>();
+	    for (Entry<Integer, Object> entry : result.entrySet()){
+	    	output.add(new Tuple2<>(entry.getKey(), (long)entry.getValue()));
+	    }
+
+	    JavaPairRDD<Integer, Long> output_rdd = sc.parallelizePairs(output);
+	    output_rdd.saveAsHadoopFile(args[2], Integer.class, String.class, TextOutputFormat.class);
+
+        sc.close();
+    }
+}

spark/src/test/java/com/matthewrathbone/sparktest/SparkJoinsTest.java

@@ -0,0 +1,76 @@
+package test.java.com.matthewrathbone.sparktest;
+
+import java.io.File;
+
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import main.java.com.matthewrathbone.sparktest.SparkJoins;
+
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import scala.Tuple2;
+
+import com.google.common.base.Optional;
+import com.google.common.io.Files;
+
+public class SparkJoinsTest implements Serializable {
+	  private transient JavaSparkContext sc;
+	  private transient File tempDir;
+
+	  @Before
+	  public void setUp() {
+	    sc = new JavaSparkContext("local", "SparkJoinsTest");
+	    tempDir = Files.createTempDir();
+	    tempDir.deleteOnExit();
+	  }
+
+	  @After
+	  public void tearDown() {
+	    sc.stop();
+	    sc = null;
+	  }
+	  
+	  @Test
+	  public void sortByKey() {
+	    List<Tuple2<Integer, Integer>> transactions = new ArrayList<>();
+	    transactions.add(new Tuple2<>(1, 1));
+	    transactions.add(new Tuple2<>(2, 1));
+	    transactions.add(new Tuple2<>(2, 1));
+	    transactions.add(new Tuple2<>(3, 2));
+	    transactions.add(new Tuple2<>(3, 1));
+	    
+	    List<Tuple2<Integer, String>> users = new ArrayList<>();
+	    users.add(new Tuple2<>(1, "US"));
+	    users.add(new Tuple2<>(2, "GB"));
+	    users.add(new Tuple2<>(3, "FR"));
+
+	    JavaPairRDD<Integer, Integer> transactions_rdd = sc.parallelizePairs(transactions);
+	    JavaPairRDD<Integer, String> users_rdd = sc.parallelizePairs(users);
+
+	    JavaRDD<Tuple2<Integer,Optional<String>>> leftJoinOutput = SparkJoins.joinData(transactions_rdd, users_rdd);
+	    
+	    Assert.assertEquals(4, leftJoinOutput.count());
+	    JavaPairRDD<Integer, String> res = SparkJoins.modifyData(leftJoinOutput);
+	    List<Tuple2<Integer, String>> sortedRes = res.sortByKey().collect();
+	    Assert.assertEquals(1, sortedRes.get(0)._1.intValue());
+	    Assert.assertEquals(1, sortedRes.get(1)._1.intValue());
+	    Assert.assertEquals(1, sortedRes.get(2)._1.intValue());
+	    Assert.assertEquals(2, sortedRes.get(3)._1.intValue());
+	    
+	    Map<Integer, Object> result = SparkJoins.countData(res);
+	    Assert.assertEquals((long)3, result.get(1));
+	    Assert.assertEquals((long)1, result.get(2));
+	    
+	  }
+}