updated made to project4

Author: imtiazahmad007

project4/src/main/java/com/jobreadyprogrammer/mappers/HouseMapper.java

@@ -36,4 +36,4 @@ public House call(Row value) throws Exception {
 
 	}
 
-}
+}

project4/src/main/java/com/jobreadyprogrammer/mappers/LineMapper.java

@@ -6,15 +6,16 @@
 import org.apache.spark.api.java.function.FlatMapFunction;
 import org.apache.spark.sql.Row;
 
-public class LineMapper implements FlatMapFunction<Row, String> {
-	
+public class LineMapper implements FlatMapFunction<Row, String>{
+
+	/**
+	 * 
+	 */
 	private static final long serialVersionUID = 1L;
-	
-	@Override
-	public Iterator<String> call(Row value) throws Exception {
 
-		return Arrays.asList(value.toString().split(" ")).iterator();
-		
+	@Override
+	public Iterator<String> call(Row row) throws Exception {
+		return Arrays.asList(row.toString().split(" ")).iterator();
 	}
-
-}
+	
+}

project4/src/main/java/com/jobreadyprogrammer/pojos/House.java

@@ -1,8 +1,14 @@
 package com.jobreadyprogrammer.pojos;
 
+import java.io.Serializable;
 import java.util.Date;
 
-public class House {
+public class House implements Serializable {
+	
+	/**
+	 * 
+	 */
+	private static final long serialVersionUID = 1L;
 
 	private int id;
 	private String address;
@@ -50,9 +56,5 @@ public Date getVacantBy() {
 	public void setVacantBy(Date vacantBy) {
 		this.vacantBy = vacantBy;
 	}
-
-
-
-	
 
-}
+}

project4/src/main/java/com/jobreadyprogrammer/pojos/Line.java

@@ -7,11 +7,12 @@ public static void main(String[] args) {
 //		ArrayToDataset app = new ArrayToDataset();
 //		app.start();
 
-//		CsvToDatasetHouseToDataframe app = new CsvToDatasetHouseToDataframe();
-//		app.start();
+		CsvToDatasetHouseToDataframe app = new CsvToDatasetHouseToDataframe();
+		app.start();
 
-		WordCount wc = new WordCount();
-		wc.start();
+//		WordCount wc = new WordCount();
+//		wc.start();
+		
 	}
 
 

project4/src/main/java/com/jobreadyprogrammer/spark/Application.java

@@ -1,11 +1,13 @@
 package com.jobreadyprogrammer.spark;
 
+import java.io.Serializable;
 import java.util.Arrays;
 import java.util.List;
 
-import org.apache.spark.sql.Encoders;
-import org.apache.spark.sql.Row;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.api.java.function.ReduceFunction;
 import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.SparkSession;
 
 public class ArrayToDataset {
@@ -16,14 +18,34 @@ public void start() {
 				.master("local")
 				.getOrCreate();
 
-		String [] stringList = new String[] {"Banana", "Glass", "Computer", "Car"};
+		String [] stringList = new String[] {"Banana", "Car", "Glass", "Banana", "Computer", "Car"};
 
 		List<String> data = Arrays.asList(stringList);
 
-		Dataset<String> ds = spark.createDataset(data, Encoders.STRING());
-
-		Dataset<Row> df = ds.toDF();
+		Dataset<String> ds =  spark.createDataset(data, Encoders.STRING());
+		
+		 ds = ds.map((MapFunction<String, String>) row -> "word: " + row, Encoders.STRING());
+		 ds.show(10);
+		 
+		String stringValue = ds.reduce(new StringReducer());
+		
+		System.out.println(stringValue);
 
 	}
 
+	
+	static class StringReducer implements ReduceFunction<String>, Serializable {
+
+		/**
+		 * 
+		 */
+		private static final long serialVersionUID = 1L;
+
+		@Override
+		public String call(String v1, String v2) throws Exception {
+			return v1 + v2;
+		}
+		
+	}
+
 }

project4/src/main/java/com/jobreadyprogrammer/spark/ArrayToDataset.java

@@ -3,11 +3,13 @@
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
+import static org.apache.spark.sql.functions.*;
 import org.apache.spark.sql.SparkSession;
 
 import com.jobreadyprogrammer.mappers.HouseMapper;
 import com.jobreadyprogrammer.pojos.House;
 
+
 public class CsvToDatasetHouseToDataframe {
 
 	public void start() {
@@ -25,28 +27,21 @@ public void start() {
 		        .option("header", true)
 		        .option("sep", ";")
 		        .load(filename);
-		 
-		    System.out.println("*** Houses ingested in a dataframe");
 
-		    df.show(5);
-		    df.printSchema();
-		 
-		    
-		    Dataset<House> houseDS = df.map(
-		        new HouseMapper(), Encoders.bean(House.class));
+		    System.out.println("House ingested in a dataframe: ");
+//		    df.show(5);
+//		    df.printSchema();
+		
+		    Dataset<House> houseDS = df.map(new HouseMapper(), Encoders.bean(House.class));
 
-		    System.out.println("below schema is for house dataset");
-		    houseDS.show(5, 17);
+		    System.out.println("*****House ingested in a dataset: *****");
+
+		    houseDS.show(5);
 		    houseDS.printSchema();
 
-		    System.out.println("below is the houses dataframe");
-		    
 		    Dataset<Row> df2 = houseDS.toDF();
-
-//		    df2.printSchema();
-//		    df2.show(5, 17);
-		    
-		
+		    df2 = df2.withColumn("formatedDate", concat(df2.col("vacantBy.date"), lit("_"), df2.col("vacantBy.year")));
+		    df2.show(10);
 	}
 
 

project4/src/main/java/com/jobreadyprogrammer/spark/CsvToDatasetHouseToDataframe.java

@@ -7,14 +7,20 @@
 
 import com.jobreadyprogrammer.mappers.LineMapper;
 
-import breeze.linalg.Options.Value;
-
 public class WordCount {
 
 	public void start() {
 
+		 String boringWords = " ('a', 'an', 'and', 'are', 'as', 'at', 'be', 'but', 'by',\r\n" + 
+			  		"'for', 'if', 'in', 'into', 'is', 'it',\r\n" + 
+			  		"'no', 'not', 'of', 'on', 'or', 'such',\r\n" + 
+			  		"'that', 'the', 'their', 'then', 'there', 'these',\r\n" + 
+			  		"'they', 'this', 'to', 'was', 'will', 'with', 'he', 'she'," + 
+			  		"'your', 'you', 'I', "
+			  		+ " 'i','[',']', '[]', 'his', 'him', 'our', 'we') ";
+		 
 		SparkSession spark = SparkSession.builder()
-		        .appName("CSV to dataframe to Dataset<House> and back")
+		        .appName("unstructured text to flatmap")
 		        .master("local")
 		        .getOrCreate();
 
@@ -23,31 +29,20 @@ public void start() {
 		Dataset<Row> df = spark.read().format("text")
 		        .load(filename);
 
-		 df.show(5);
-		 df.printSchema();
-		 
-		  Dataset<String> lineDS = df.flatMap(
-			        new LineMapper(), Encoders.STRING());
+//		df.printSchema();
+//		df.show(10);
+		
+		Dataset<String> wordsDS = df.flatMap(new LineMapper(), Encoders.STRING());
+
+		Dataset<Row> df2 = wordsDS.toDF();
+		
+		df2 = df2.groupBy("value").count();
+		df2 = df2.orderBy(df2.col("count").desc());
+		df2 = df2.filter("lower(value) NOT IN " + boringWords);
+		
+		df2.show(500);
 
-		 
-		  lineDS.printSchema();
-		  lineDS.show(10, 200);
-		  
-		  String boringWords = "(   'a', 'an', 'and', 'are', 'as', 'at', 'be', 'but', 'by',\r\n" + 
-		  		"      'for', 'if', 'in', 'into', 'is', 'it',\r\n" + 
-		  		"      'no', 'not', 'of', 'on', 'or', 'such',\r\n" + 
-		  		"      'that', 'the', 'their', 'then', 'there', 'these',\r\n" + 
-		  		"      'they', 'this', 'to', 'was', 'will', 'with', 'he', 'she', 'your', 'you', 'I', "
-		  		+ " 'i','[',']', '[]', 'his', 'him', 'our', 'we')";
-		  
-		  Dataset<Row> df2 = lineDS.toDF();
-		  df2 = df2.groupBy("value").count();
-		  df2 = df2.filter("lower(value) NOT IN" + boringWords);
-		  df2 = df2.orderBy(df2.col("count").desc());
-		  
 
-		  df2.printSchema();
-		  df2.show(100);
 	}