Spark 01: Movie Rating Counter

slgobinath · slgobinath · commit 38ab03b96f7f · 2019-01-24T14:53:56.000-05:00
diff --git a/spark/Movie Ratings Counter/build.sbt b/spark/Movie Ratings Counter/build.sbt
@@ -0,0 +1,7 @@
+name := "Movie Ratings Counter"
+
+version := "0.1"
+
+scalaVersion := "2.12.8"
+
+libraryDependencies += "org.apache.spark" %% "spark-core" % "2.4.0"
diff --git a/spark/Movie Ratings Counter/project/build.properties b/spark/Movie Ratings Counter/project/build.properties
@@ -0,0 +1 @@
+sbt.version = 1.2.8
diff --git a/spark/Movie Ratings Counter/src/main/scala/com/javahelps/spark/MovieRatingsCounter.scala b/spark/Movie Ratings Counter/src/main/scala/com/javahelps/spark/MovieRatingsCounter.scala
@@ -0,0 +1,32 @@
+package com.javahelps.spark
+
+import org.apache.log4j.{Level, Logger}
+import org.apache.spark.SparkContext
+
+
+object MovieRatingsCounter {
+
+    def main(args: Array[String]): Unit = {
+
+        Logger.getLogger("org").setLevel(Level.ERROR)
+
+        val sc = new SparkContext("local[*]", "MovieRatingsCounter")
+
+        // Read a text file
+        var data = sc.textFile("/tmp/ml-latest-small/ratings.csv")
+
+        // Extract the first row which is the header
+        val header = data.first();
+
+        // Filter out the header from the dataset
+        data = data.filter(row => row != header)
+
+        val result = data.map(line => line.split(',')(2).toFloat) // Extract rating from line as float
+          .countByValue() // Count number of occurrences of each number
+
+        // Sort and print the result
+        result.toSeq
+          .sorted
+          .foreach(println)
+    }
+}
