task4

Author: mtumilowicz

src/main/scala/app/task1/Task1.scala

@@ -5,7 +5,7 @@ import org.apache.spark.sql.{Dataset, SparkSession}
 object Task1 extends App {
 
   implicit val spark: SparkSession = bootstrapSpark()
-  val file = loadFile("Task1.txt").cache()
+  val file = loadFile("Dataset.txt").cache()
   val fileAnalysis = new TextFileAnalysis(file)
   val numAs = fileAnalysis.lineContaining("a").count()
   val numBs = fileAnalysis.lineContaining("b").count()

src/main/scala/app/task4/Task4.scala

@@ -1,41 +1,42 @@
-//package app.task4
-//
-//import org.apache.spark.sql.{DataFrame, SparkSession}
-//
-//object Task4 {
-//  def main(args: Array[String]) {
-//    implicit val spark: SparkSession = bootstrapSpark()
-//    val address: DataFrame = loadJsonFile("Task4_2")
-//
-//    val customerInfo: DataFrame = loadCsvFile("Task4_1.csv")
-//
-//    val enrichedAddress = address.join(customerInfo.as('info), customerInfo("CustomerId") === address("CustomerId"))
-//      .select()
-//
-//    investigate(enrichedAddress)
-//
-//    spark.stop()
-//  }
-//
-//  def loadJsonFile(filePath: String)(implicit spark: SparkSession): DataFrame =
-//    spark.read.json(filePath)
-//
-//  def loadCsvFile(filePath: String)(implicit spark: SparkSession): DataFrame =
-//    spark.read.option("header", "true").csv(filePath)
-//
-//  def investigate(dataFrame: DataFrame): Unit = {
-//    dataFrame.show()
-//    dataFrame.printSchema()
-//  }
-//
-//  def bootstrapSpark(): SparkSession = {
-//    val spark = SparkSession.builder
-//      .appName("Simple Application")
-//      .master("local")
-//      .getOrCreate()
-//
-//    spark.sparkContext.setLogLevel("ERROR")
-//
-//    spark
-//  }
-//}
+package app.task4
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+object Task4 {
+  def main(args: Array[String]) {
+
+    implicit val spark: SparkSession = bootstrapSpark()
+
+    val address: DataFrame = loadJsonFile("task4/Dataset2")
+
+    val customerInfo: DataFrame = loadCsvFile("task4/Dataset1.csv")
+
+    val enrichedAddress = address.join(customerInfo, "CustomerId")
+
+    investigate(enrichedAddress)
+
+    spark.stop()
+  }
+
+  def loadJsonFile(filePath: String)(implicit spark: SparkSession): DataFrame =
+    spark.read.json(filePath)
+
+  def loadCsvFile(filePath: String)(implicit spark: SparkSession): DataFrame =
+    spark.read.option("header", "true").csv(filePath)
+
+  def investigate(dataFrame: DataFrame): Unit = {
+    dataFrame.show()
+    dataFrame.printSchema()
+  }
+
+  def bootstrapSpark(): SparkSession = {
+    val spark = SparkSession.builder
+      .appName("Simple Application")
+      .master("local")
+      .getOrCreate()
+
+    spark.sparkContext.setLogLevel("ERROR")
+
+    spark
+  }
+}