Feature/scala code/ch07 deepak (#17)

* ch07-DataframeRedisWriter

* ch07:DatasourceMongodbWriter

* ch07:DatasourceMongodbReader

* ch07-DatasourceTextfileWriter

* ch07-DatasourceTextfileReader

Author: deepakmca05

code/chap07/scala/src/main/scala/org/data/algorithms/spark/ch07/DatasourceTextfileReader.scala

@@ -0,0 +1,54 @@
+package org.data.algorithms.spark.ch07
+
+import org.apache.spark.sql.SparkSession
+
+import scala.io.Source
+
+object DatasourceTextfileReader {
+
+  def debugFile(inputPath: String) = {
+    val bufferedSource = Source.fromFile(inputPath)
+    for (line <- bufferedSource)
+      println(line)
+    bufferedSource.close()
+  }
+
+  def main(args: Array[String]): Unit = {
+    if (args.length != 1) {
+      System.err.println("Usage: DatasourceTextfileReader <output-path>")
+      System.exit(-1)
+    }
+    //create an instance of SparkSession
+    val spark =
+      SparkSession.
+        builder().
+        master("local[*]").
+        getOrCreate()
+
+    // read name of input file
+    val inputPath = args(0)
+    println(s"inputPath: ${inputPath}")
+    debugFile(inputPath)
+    /*
+    ================================================
+    # Create an RDD[String] from a given Text File
+    ================================================
+    */
+    val records = spark.sparkContext.textFile(inputPath)
+    println(s"records = ${records}")
+    println(s"records.count() = ${records.count()}")
+    println(s"records.collect() = ${records.collect().mkString("[",",","]")}")
+    /*
+    #================================================
+    # Transform an RDD[String] to RDD[Integer]
+    #================================================
+    */
+    val numbers = records.flatMap(rec =>  rec.split(",")).map(_.toInt)
+    println(s"numbers = ${numbers}")
+    println(s"numbers.count() = ${numbers.count()}")
+    println(s"numbers.collect() = ${numbers.collect().mkString("[",",","]")}")
+    //Done.
+    spark.stop()
+  }
+
+}

code/chap07/scala/src/main/scala/org/data/algorithms/spark/ch07/DatasourceTextfileWriter.scala

@@ -0,0 +1,53 @@
+package org.data.algorithms.spark.ch07
+
+import org.apache.spark.sql.SparkSession
+
+object DatasourceTextfileWriter {
+  def main(args: Array[String]): Unit = {
+    if (args.length != 1) {
+      System.err.println("Usage: DatasourceTextfileWriter <output-path>")
+      System.exit(-1)
+    }
+    //create an instance of SparkSession
+    val spark =
+      SparkSession.
+        builder().
+        master("local[*]").
+        getOrCreate()
+
+    // read name of input file
+    val outputPath = args(0)
+    println(s"outputPath: ${outputPath}")
+    /*
+    ================================================
+    # Create an RDD[String]
+    ================================================
+    */
+    val data = List("data element 1", "data element 2", "data element 3", "data element 4")
+    println(s"data = ${data}")
+    val records = spark.sparkContext.parallelize(data)
+    println(s"records = ${records}")
+    println(s"records.count() = ${records.count()}")
+    println(s"records.collect() = ${records.collect().mkString("[",",","]")}")
+    /*
+    #================================================
+    # Save an RDD[String] to an output path
+    #================================================
+    */
+    records.saveAsTextFile(outputPath)
+    /*
+    #================================================
+    # read back from an output path and create and RDD[String]
+      #================================================
+    #
+    */
+    val loadedRecords = spark.sparkContext.textFile(outputPath)
+    println(s"loaded_records = ${loadedRecords}")
+    println(s"loaded_records.count() = ${loadedRecords.count()}")
+    println(s"loaded_records.collect() = ${loadedRecords.collect().mkString("[",",","]")}")
+
+    //Done.
+    spark.stop()
+  }
+
+}