Removed File wrapper aroung input path to read files from local and s3

Author: mahen-n

README.md

@@ -31,9 +31,9 @@ A Spark bootstrap project written in Scala with gradle as build tool.
 
 #### Run sparkSubmit task
 
-- Runs a `spark-submit` with class `dev.template.spark.RddCollect`
+Gradle sparkSubmit task is configured to run with class the `dev.template.spark.RddCollect`
 
-      ./gradlew sparkSubmit
+	./gradlew sparkSubmit
 
 #### Spark Submit commands in shell
 

gradle.properties

@@ -5,6 +5,7 @@ commonsIO=2.13.0
 deltaVersion=2.4.0
 #kafka
 confluentVersion=7.5.0
+kafkaClientVersion=3.4.0
 #logging
 slf4jVersion=1.7.21
 logbackVersion=1.1.7

src/main/scala/dev/template/spark/CovidDataPartitioner.scala

@@ -4,8 +4,6 @@ import dev.template.spark.sink.Writer
 import dev.template.spark.source.Reader
 import org.apache.spark.sql.{Dataset, Row, SaveMode, SparkSession}
 
-import java.io.File
-
 object CovidDataPartitioner
     extends App
     with SparkSessionWrapper
@@ -26,7 +24,6 @@ object CovidDataPartitioner
              |  fips,
              |  cases,
              |  deaths from covid
-             |  group by all
              |
              |""".stripMargin)
       .cache()
@@ -54,8 +51,12 @@ object CovidDataPartitioner
     throw new RuntimeException("Requires input file us-counties-recent.csv")
   }
 
-  private val inputFilePath = new File(args(0)).toString
-  private val outputPath = new File(args(1)).toString
+  var inputFilePath = args(0)
+  var outputPath: String = args(1)
+
+  log.info("Input path " + inputFilePath)
+  log.info("Output path " + outputPath)
+
   writeParquet(spark, inputFilePath, outputPath)
 
 }

src/main/scala/dev/template/spark/Main.scala

@@ -44,7 +44,7 @@ object Main extends App {
               |""".stripMargin)
     throw new RuntimeException("Requires input file people-example.csv")
   }
-  private val inputFilePath = new File(args(0)).toString
+  private val inputFilePath = args(0)
   val calculateAverageAge = new CalculateAverageAge()
   calculateAverageAge.calculateAverageAge(inputFilePath)
 }

src/main/scala/dev/template/spark/source/Reader.scala

@@ -17,10 +17,28 @@ trait Reader {
     .option("inferSchema", true)
     .option("mode", "DROPMALFORMED")
 
-  def readKafka(spark: SparkSession, topic: String, options: Map[String, String]) = spark
-    .read
-    .format("kafka")
-    .options(options)
-    .option("subscribe", topic)
-    .load()
+  def readDelta(spark: SparkSession, path: String, options: Map[String, String] = Map()) =
+    spark.read.format("delta").options(options).load(path)
+
+  /**
+   * Kafka reader requires kafka consumer properties.
+   *
+   * @param spark
+   *   spark session
+   * @param topic
+   *   kafka topic to consume
+   * @param kafkaConfig
+   *   Kafka consumer properties
+   * @return
+   */
+  def readKafka(spark: SparkSession, topic: String, kafkaConfig: Map[String, String] = Map()) =
+    spark
+      .read
+      .format("kafka")
+      .options(kafkaConfig)
+      .option("subscribe", topic)
+      .option("startingOffsets", "earliest")
+      .option("endingOffsets", "latest")
+      .load()
+
 }

src/test/scala/dev/template/spark/CovidDataPartitionerTest.scala

@@ -21,14 +21,13 @@ class CovidDataPartitionerTest extends AnyFunSpec with SparkSessionTestWrapper {
 
     it("number of reported_date partitions should be 30") {
 
-
       val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
       val reportedDateFolderCount = fs
         .listStatus(new Path(outputPath),
-          new PathFilter {
-            override def accept(path: Path): Boolean =
-              path.getName.contains("reported_date")
-          })
+                    new PathFilter {
+                      override def accept(path: Path): Boolean =
+                        path.getName.contains("reported_date")
+                    })
         .length
 
       assertEquals(30, reportedDateFolderCount)