Initial Commit

Author: LearningJournal

17-RankingDemo/build.sbt

@@ -0,0 +1,14 @@
+name := "RankingDemo"
+organization := "guru.learningjournal"
+version := "0.1"
+scalaVersion := "2.12.10"
+
+autoScalaLibrary := false
+val sparkVersion = "3.0.0-preview2"
+
+val sparkDependencies = Seq(
+  "org.apache.spark" %% "spark-core" % sparkVersion,
+  "org.apache.spark" %% "spark-sql" % sparkVersion
+)
+
+libraryDependencies ++= sparkDependencies

17-RankingDemo/data/summary.parquet

@@ -0,0 +1,39 @@
+# Set everything to be logged to the console
+log4j.rootCategory=WARN, console
+
+# define console appender
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.target=System.out
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+
+#application log
+log4j.logger.guru.learningjournal.spark.examples=INFO, console, file
+log4j.additivity.guru.learningjournal.spark.examples=false
+
+#define rolling file appender
+log4j.appender.file=org.apache.log4j.RollingFileAppender
+log4j.appender.file.File=${spark.yarn.app.container.log.dir}/${logfile.name}.log
+#define following in Java System
+# -Dlog4j.configuration=file:log4j.properties
+# -Dlogfile.name=hello-spark
+# -Dspark.yarn.app.container.log.dir=app-logs
+log4j.appender.file.ImmediateFlush=true
+log4j.appender.file.Append=false
+log4j.appender.file.MaxFileSize=500MB
+log4j.appender.file.MaxBackupIndex=2
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.conversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+
+
+# Recommendations from Spark template
+log4j.logger.org.apache.spark.repl.Main=WARN
+log4j.logger.org.spark_project.jetty=WARN
+log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR
+log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO
+log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO
+log4j.logger.org.apache.parquet=ERROR
+log4j.logger.parquet=ERROR
+log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
+log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR
+

17-RankingDemo/log4j.properties

@@ -0,0 +1,33 @@
+package guru.learningjournal.spark.examples
+
+import org.apache.log4j.Logger
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.expressions.Window
+import org.apache.spark.sql.functions._
+
+object RankingDemo extends Serializable {
+  @transient lazy val logger: Logger = Logger.getLogger(getClass.getName)
+
+  def main(args: Array[String]): Unit = {
+
+    val spark = SparkSession.builder()
+      .appName("Ranking Demo")
+      .master("local[3]")
+      .getOrCreate()
+
+    val summaryDF = spark.read.parquet("data/summary.parquet")
+
+    summaryDF.sort("Country", "WeekNumber").show()
+
+    val rankWindow = Window.partitionBy("Country")
+      .orderBy(col("InvoiceValue").desc)
+      .rowsBetween(Window.unboundedPreceding, Window.currentRow)
+
+    summaryDF.withColumn("Rank", dense_rank().over(rankWindow))
+      .where(col("Rank") ===1)
+      .sort("Country", "WeekNumber")
+      .show()
+
+    spark.stop()
+  }
+}