L01-Project_Big Data Analytics with Hadoop and Apache Spark.ipynb

{"cells":[{"cell_type":"markdown","source":["# Student Scores Project | Big Data Analytics w/ Hadoop and Apache Spark\n\n<img src = \"https://hadoopinrealworld.com/wp-content/uploads/2017/09/Spark-vs-Hadoop-Comparison-Chart.png\" width = \"500\">\n\nThis project shows the combined capabilities of Hadoop and Apache Spark for a project showing some statistics on a student score data set. The practice of combining the strong sides of these two projects (i.e., Hadoop HDFS + Apache Spark) is well accepted by the data teams in those days.\n\n### What is Hadoop?\n\nThe Apache Hadoop Project is an open source project and consists of four main modules:\n\n*  HDFS – Hadoop Distributed File System.\n*  MapReduce. The processing component of the Hadoop ecosystem. It scales horizontally and is slow as it uses data storage. In the last decade, professionals use Apache Spark or Flink instead of MapReduce.\n*  YARN. Yet Another Resource Negotiator. It is used for computing resources and job scheduling.\n*  Hadoop Common. Also called as Hadoop core providing support for all other Hadoop components.\n\nAmong these modules, this project focuses on Hadoop HDFS. It is the file system managing the storage of large data sets. It can handle both structured and unstructured data. Hadoop stores the data to disks using HDFS.\n\n### What is Apache Spark?\n\nApache Spark is an open source project. It uses RAM for caching and processing data ans is designed for fast performance. Resilient Distributed Dataset (RDD) is the data structure of Spark.  It consists of five main components:\n\n*  Apache Spark Core. Spark Core is the basis and includes functions of *scheduling, task dispatching, input and output operations, etc.* \n*  Spark Streaming. It is used for processing of live data streams with data sources of Kafka, Kinesis, Flume, etc.\n*  Spark SQL uses this component to gather information about the structured data and how the data is processed.\n*  Machine Learning Library (MLlib) includes machine learning algorithms.\n*  GraphX is for facilitating graph analytics tasks.\n\nIn this notebook, we use Apache Spark Core functions as it uses memory to speed up the computations.  \n\n## Dataset ▶\n\nDataset consists of student scores of different subjects in CSV file format. There are 40 rows of data and one header. It is relatively small for the sake of practicing functions. In big data world, data size can go up to petabytes. 4 columns are given below:\n\nStudent | Subject |\tClass Score |\tTest Score\n\n\n*Please note that: the project herein is inspired by the lecture notes of \"Big Data Analytics with Hadoop and Apache Spark\" by Kumaran Ponnambalam.*\n\n## Target 🎯\n\nThis notebook will walk you through the (1) data loading into HDFS format and (2) data processing with Spark. Here's the outline:\n\n* Import functions\n* Data load \n    * Parquet File + Gzip codec  \n    We prefer Parquet in HDFS as it reads col by col, provides schema, is compressible and splittable. It is ideal for analytics  \n    We prefer gzip codec as it provides good compression but is not splittable and provides moderate performance. It is good for analytical purposes.  \n    * Schema optimization\n* Data processing\n    * Computing total score\n    * Printing total score for physics\n    * Computing avg total score\n    * Finding student with highest score"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"85dd9556-2bfc-42f6-92c2-bb8ba7d5516c","inputWidgets":{},"title":""}}},{"cell_type":"markdown","source":["# 1 Data Loading 📊\n\nReading CSV file, partioning in HDFS format and reading parquet file 🤯"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"339ef157-f11c-4052-8f72-af834c7990dd","inputWidgets":{},"title":""}}},{"cell_type":"code","source":["%scala\n//1 - Reading csv file\nval filepath = \"dbfs:/FileStore/FileStore/st_scores.csv\"\n\n//read the raw CSV file - a spark DataFrame\nval raw_stdata = spark.read.format(\"csv\")\n                      .option(\"inferSchema\", \"true\")\n                      .option(\"header\", \"true\")\n                      .load(filepath)\n                      .withColumnRenamed(\"Class Score\",\"ClassScore\")\n                      .withColumnRenamed(\"Test Score\",\"TestScore\");\n\n//checking schema - ensuring whether everything is gone good or not\nraw_stdata.printSchema()\n\n//checking data\nraw_stdata.show(5)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"52f381ec-9499-4a36-8399-8a283b8aaa46","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[{"name":"raw_stdata","typeStr":"org.apache.spark.sql.DataFrame","schema":{"type":"struct","fields":[{"name":"Student","type":"string","nullable":true,"metadata":{}},{"name":"Subject","type":"string","nullable":true,"metadata":{}},{"name":"ClassScore","type":"integer","nullable":true,"metadata":{}},{"name":"TestScore","type":"double","nullable":true,"metadata":{}},{"name":"_c4","type":"string","nullable":true,"metadata":{}},{"name":"_c5","type":"string","nullable":true,"metadata":{}},{"name":"_c6","type":"string","nullable":true,"metadata":{}}]},"tableIdentifier":null}],"data":"<div class=\"ansiout\">root\n |-- Student: string (nullable = true)\n |-- Subject: string (nullable = true)\n |-- ClassScore: integer (nullable = true)\n |-- TestScore: double (nullable = true)\n |-- _c4: string (nullable = true)\n |-- _c5: string (nullable = true)\n |-- _c6: string (nullable = true)\n\n+-------+---------+----------+---------+----+----+----+\n|Student|  Subject|ClassScore|TestScore| _c4| _c5| _c6|\n+-------+---------+----------+---------+----+----+----+\n|  James|     Math|        95|   65.175|null|null|null|\n|  James|Chemistry|        50|    32.45|null|null|null|\n|  James|  Physics|        48|   37.675|null|null|null|\n|  James|  Biology|        75|   76.725|null|null|null|\n|   Lora|     Math|        45|   49.225|null|null|null|\n+-------+---------+----------+---------+----+----+----+\nonly showing top 5 rows\n\nfilepath: String = dbfs:/FileStore/FileStore/st_scores.csv\nraw_stdata: org.apache.spark.sql.DataFrame = [Student: string, Subject: string ... 5 more fields]\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">root\n-- Student: string (nullable = true)\n-- Subject: string (nullable = true)\n-- ClassScore: integer (nullable = true)\n-- TestScore: double (nullable = true)\n-- _c4: string (nullable = true)\n-- _c5: string (nullable = true)\n-- _c6: string (nullable = true)\n\n+-------+---------+----------+---------+----+----+----+\nStudent|  Subject|ClassScore|TestScore| _c4| _c5| _c6|\n+-------+---------+----------+---------+----+----+----+\n  James|     Math|        95|   65.175|null|null|null|\n  James|Chemistry|        50|    32.45|null|null|null|\n  James|  Physics|        48|   37.675|null|null|null|\n  James|  Biology|        75|   76.725|null|null|null|\n   Lora|     Math|        45|   49.225|null|null|null|\n+-------+---------+----------+---------+----+----+----+\nonly showing top 5 rows\n\nfilepath: String = dbfs:/FileStore/FileStore/st_scores.csv\nraw_stdata: org.apache.spark.sql.DataFrame = [Student: string, Subject: string ... 5 more fields]\n</div>"]}}],"execution_count":0},{"cell_type":"code","source":["%scala\n// 2 - Creating partitioned HDFS store\n// storing CSV file as Parquet (read col by col, schema, compressible and splittable +> ideal for analytics) in HDFS to improve the performance\n// w/  gzip compression: good compression + not splittable + moderate performance => good for analytical purposes\n// then partitioning by Subject - provides limitied list of partitions\nval fileout = \"dbfs:/FileStore/FileStore/partitioned_st\"\n\nraw_stdata.write\n          .format(\"parquet\")\n          .mode(\"overwrite\")\n          .option(\"compression\", \"gzip\")\n          .partitionBy(\"Subject\")\n          .save(fileout)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"70722db3-f0a2-4f27-b242-cba2a588c0c7","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"<div class=\"ansiout\">fileout: String = dbfs:/FileStore/FileStore/partitioned_st\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">fileout: String = dbfs:/FileStore/FileStore/partitioned_st\n</div>"]}}],"execution_count":0},{"cell_type":"code","source":["%scala\n// 3 - reading partitioned data into Data Frame\n\nval st_data = spark.read\n                   .parquet(fileout)\n\nprintln(\"# of partitions in dataset : \" + st_data.rdd.getNumPartitions)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"a6fd0a8c-e4f1-4f38-89ce-23a03d36e8ae","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[{"name":"st_data","typeStr":"org.apache.spark.sql.DataFrame","schema":{"type":"struct","fields":[{"name":"Student","type":"string","nullable":true,"metadata":{}},{"name":"ClassScore","type":"integer","nullable":true,"metadata":{}},{"name":"TestScore","type":"double","nullable":true,"metadata":{}},{"name":"_c4","type":"string","nullable":true,"metadata":{}},{"name":"_c5","type":"string","nullable":true,"metadata":{}},{"name":"_c6","type":"string","nullable":true,"metadata":{}},{"name":"Subject","type":"string","nullable":true,"metadata":{}}]},"tableIdentifier":null}],"data":"<div class=\"ansiout\"># of partitions in dataset : 4\nst_data: org.apache.spark.sql.DataFrame = [Student: string, ClassScore: int ... 5 more fields]\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"># of partitions in dataset : 4\nst_data: org.apache.spark.sql.DataFrame = [Student: string, ClassScore: int ... 5 more fields]\n</div>"]}}],"execution_count":0},{"cell_type":"markdown","source":["The HDFS file is created correctly on the gicen path above. This data is then read into a partitioned data frame. The number of partitions in the data frame is printed as 4. It iis the default parallelism for this installation on DataBricks. It can be customized. \n\nIn the following section, let's perform the data analytics."],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"c399adef-bfcb-4773-8257-1311816e6c10","inputWidgets":{},"title":""}}},{"cell_type":"markdown","source":["# 2 - Data Processing 🔧\n\nComputing total scores, average score, top score, etc...\n\n## 2.1- Let's find out total score: class score + test score"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"d672a293-423b-419a-8cc2-0bca5412c638","inputWidgets":{},"title":""}}},{"cell_type":"code","source":["%scala\n// 1- Let's find out total score: class score + test score\n// Transform and Map operations to add columns\n\nval tot_score = st_data.withColumn(\"TotScore\",\n                        st_data.col(\"ClassScore\")\n                        + st_data.col(\"TestScore\"))\n                        \ntot_score.show(5)\n\nprintln(\"--------Explain--------\")\ntot_score.explain\nprintln(\"--------End of Explain--------\")\nprintln(\"# of partitions in dataset : \" + tot_score.rdd.getNumPartitions)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"f626ceef-6d2e-44ec-90f7-677b5d484a55","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[{"name":"tot_score","typeStr":"org.apache.spark.sql.DataFrame","schema":{"type":"struct","fields":[{"name":"Student","type":"string","nullable":true,"metadata":{}},{"name":"ClassScore","type":"integer","nullable":true,"metadata":{}},{"name":"TestScore","type":"double","nullable":true,"metadata":{}},{"name":"_c4","type":"string","nullable":true,"metadata":{}},{"name":"_c5","type":"string","nullable":true,"metadata":{}},{"name":"_c6","type":"string","nullable":true,"metadata":{}},{"name":"Subject","type":"string","nullable":true,"metadata":{}},{"name":"TotScore","type":"double","nullable":true,"metadata":{}}]},"tableIdentifier":null}],"data":"<div class=\"ansiout\">+-------+----------+---------+----+----+----+-------+--------+\n|Student|ClassScore|TestScore| _c4| _c5| _c6|Subject|TotScore|\n+-------+----------+---------+----+----+----+-------+--------+\n|  James|        95|   65.175|null|null|null|   Math| 160.175|\n|   Lora|        45|   49.225|null|null|null|   Math|  94.225|\n|   Leny|        36|   65.175|null|null|null|   Math| 101.175|\n|   Lisa|        33|    78.65|null|null|null|   Math|  111.65|\n|  Elvis|        27|     33.0|null|null|null|   Math|    60.0|\n+-------+----------+---------+----+----+----+-------+--------+\nonly showing top 5 rows\n\n--------Explain--------\n== Physical Plan ==\nInMemoryTableScan [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], false\n   +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n         +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n            +- *(1) ColumnarToRow\n               +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 4\ntot_score: org.apache.spark.sql.DataFrame = [Student: string, ClassScore: int ... 6 more fields]\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">+-------+----------+---------+----+----+----+-------+--------+\nStudent|ClassScore|TestScore| _c4| _c5| _c6|Subject|TotScore|\n+-------+----------+---------+----+----+----+-------+--------+\n  James|        95|   65.175|null|null|null|   Math| 160.175|\n   Lora|        45|   49.225|null|null|null|   Math|  94.225|\n   Leny|        36|   65.175|null|null|null|   Math| 101.175|\n   Lisa|        33|    78.65|null|null|null|   Math|  111.65|\n  Elvis|        27|     33.0|null|null|null|   Math|    60.0|\n+-------+----------+---------+----+----+----+-------+--------+\nonly showing top 5 rows\n\n--------Explain--------\n== Physical Plan ==\nInMemoryTableScan [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], false\n   +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n         +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n            +- *(1) ColumnarToRow\n               +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 4\ntot_score: org.apache.spark.sql.DataFrame = [Student: string, ClassScore: int ... 6 more fields]\n</div>"]}}],"execution_count":0},{"cell_type":"markdown","source":["**Observation**  \nLet's take a look at the execution plan to understand how the spark function was executed. Alternatively, click on view jobs and then DAG visualization to understand the map operation. We can see that this is a simple map operation. In this case, we have   \n<1- FileScanRDD => 2- MapPartitionsRDD > as a part of scan parquet  \n<3- MapPartitionsRDD> as a part of whole stage codegen \n\n## 2.2- Let's print out total score for pyhsics"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"f54cda29-a4ca-40df-9160-52492fe5b278","inputWidgets":{},"title":""}}},{"cell_type":"code","source":["%scala\n// 2- Let's print out total score for pyhsics\n// Filter use case which should trigger a filter pushdown for the subject\n\nval physics_score = tot_score.filter($\"Subject\" === \"Physics\")\n                        \nphysics_score.show()\n\nprintln(\"--------Explain--------\")\nphysics_score.explain\nprintln(\"--------End of Explain--------\")\nprintln(\"# of partitions in dataset : \" + physics_score.rdd.getNumPartitions)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"d9c2c43b-87dd-4ce9-bcaf-31e80b4a01d4","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[{"name":"physics_score","typeStr":"org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]","schema":{"type":"struct","fields":[{"name":"Student","type":"string","nullable":true,"metadata":{}},{"name":"ClassScore","type":"integer","nullable":true,"metadata":{}},{"name":"TestScore","type":"double","nullable":true,"metadata":{}},{"name":"_c4","type":"string","nullable":true,"metadata":{}},{"name":"_c5","type":"string","nullable":true,"metadata":{}},{"name":"_c6","type":"string","nullable":true,"metadata":{}},{"name":"Subject","type":"string","nullable":true,"metadata":{}},{"name":"TotScore","type":"double","nullable":true,"metadata":{}}]},"tableIdentifier":null}],"data":"<div class=\"ansiout\">+-------+----------+---------+----+----+----+-------+---------+\n|Student|ClassScore|TestScore| _c4| _c5| _c6|Subject| TotScore|\n+-------+----------+---------+----+----+----+-------+---------+\n|  James|        48|   37.675|null|null|null|Physics|   85.675|\n|   Lora|        34|     74.8|null|null|null|Physics|    108.8|\n|   Leny|        93|   79.475|null|null|null|Physics|  172.475|\n|   Lisa|        42|    64.35|null|null|null|Physics|   106.35|\n|  Elvis|        82|     77.0|null|null|null|Physics|    159.0|\n|Micheal|        48| 31.68125|null|null|null|Physics| 79.68125|\n| Daniel|        34|     62.9|null|null|null|Physics|     96.9|\n|   Dave|        93| 66.83125|null|null|null|Physics|159.83125|\n|   Roby|        42|  54.1125|null|null|null|Physics|  96.1125|\n| Pamela|        82|    64.75|null|null|null|Physics|   146.75|\n+-------+----------+---------+----+----+----+-------+---------+\n\n--------Explain--------\n== Physical Plan ==\n*(1) Filter (isnotnull(Subject#3420) AND (Subject#3420 = Physics))\n+- InMemoryTableScan [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], [isnotnull(Subject#3420), (Subject#3420 = Physics)], false\n      +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n            +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n               +- *(1) ColumnarToRow\n                  +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 4\nphysics_score: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [Student: string, ClassScore: int ... 6 more fields]\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">+-------+----------+---------+----+----+----+-------+---------+\nStudent|ClassScore|TestScore| _c4| _c5| _c6|Subject| TotScore|\n+-------+----------+---------+----+----+----+-------+---------+\n  James|        48|   37.675|null|null|null|Physics|   85.675|\n   Lora|        34|     74.8|null|null|null|Physics|    108.8|\n   Leny|        93|   79.475|null|null|null|Physics|  172.475|\n   Lisa|        42|    64.35|null|null|null|Physics|   106.35|\n  Elvis|        82|     77.0|null|null|null|Physics|    159.0|\nMicheal|        48| 31.68125|null|null|null|Physics| 79.68125|\n Daniel|        34|     62.9|null|null|null|Physics|     96.9|\n   Dave|        93| 66.83125|null|null|null|Physics|159.83125|\n   Roby|        42|  54.1125|null|null|null|Physics|  96.1125|\n Pamela|        82|    64.75|null|null|null|Physics|   146.75|\n+-------+----------+---------+----+----+----+-------+---------+\n\n--------Explain--------\n== Physical Plan ==\n*(1) Filter (isnotnull(Subject#3420) AND (Subject#3420 = Physics))\n+- InMemoryTableScan [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], [isnotnull(Subject#3420), (Subject#3420 = Physics)], false\n      +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n            +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n               +- *(1) ColumnarToRow\n                  +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 4\nphysics_score: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [Student: string, ClassScore: int ... 6 more fields]\n</div>"]}}],"execution_count":0},{"cell_type":"markdown","source":["**Observations**  \nA simple filter is executed on the subject column and it gives the scores for Physics for all. On the execution plan, we can see that this filter was pushed down to HDFS and a single partition was read.  \nIt reads the data source once again to calculate the scores without **caching**.   \nIn the following part, we will employ **cache** for data analytics.\n\n## 2.3 Computing average total score with caching\n\nIn this part, we will also use aggregation with Spark"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"d949520a-5dfe-4209-a872-1d9c06bae407","inputWidgets":{},"title":""}}},{"cell_type":"code","source":["%scala\n// 3- caching and using aggregation functions : groupby and avg\ntot_score.cache()\n\n// group by keys\nval avgscore = tot_score.groupBy(\"Student\")\n                  .avg(\"TotScore\")\n\navgscore.show()\n\nprintln(\"--------Explain--------\")\navgscore.explain\nprintln(\"--------End of Explain--------\")\nprintln(\"# of partitions in dataset : \" + avgscore.rdd.getNumPartitions)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"7da03eaf-6635-44b4-821e-63e6d95161e8","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[{"name":"avgscore","typeStr":"org.apache.spark.sql.DataFrame","schema":{"type":"struct","fields":[{"name":"Student","type":"string","nullable":true,"metadata":{}},{"name":"avg(TotScore)","type":"double","nullable":true,"metadata":{}}]},"tableIdentifier":null}],"data":"<div class=\"ansiout\">+-------+------------------+\n|Student|     avg(TotScore)|\n+-------+------------------+\n|  James|         120.00625|\n|   Dave|       121.0359375|\n|  Elvis|113.60624999999999|\n| Pamela|       103.2484375|\n|   Lora| 99.23124999999999|\n|   Roby|         89.478125|\n|   Leny|         131.30625|\n| Daniel| 90.40468750000001|\n|Micheal|         115.45625|\n|   Lisa| 97.98750000000001|\n+-------+------------------+\n\n--------Explain--------\n== Physical Plan ==\nAdaptiveSparkPlan isFinalPlan=false\n+- HashAggregate(keys=[Student#3414], functions=[finalmerge_avg(merge sum#4487, count#4488L) AS avg(TotScore#3430)#4395])\n   +- Exchange hashpartitioning(Student#3414, 200), ENSURE_REQUIREMENTS, [plan_id=1237]\n      +- HashAggregate(keys=[Student#3414], functions=[partial_avg(TotScore#3430) AS (sum#4487, count#4488L)])\n         +- InMemoryTableScan [Student#3414, TotScore#3430], false\n               +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n                     +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n                        +- *(1) ColumnarToRow\n                           +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 1\navgscore: org.apache.spark.sql.DataFrame = [Student: string, avg(TotScore): double]\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">+-------+------------------+\nStudent|     avg(TotScore)|\n+-------+------------------+\n  James|         120.00625|\n   Dave|       121.0359375|\n  Elvis|113.60624999999999|\n Pamela|       103.2484375|\n   Lora| 99.23124999999999|\n   Roby|         89.478125|\n   Leny|         131.30625|\n Daniel| 90.40468750000001|\nMicheal|         115.45625|\n   Lisa| 97.98750000000001|\n+-------+------------------+\n\n--------Explain--------\n== Physical Plan ==\nAdaptiveSparkPlan isFinalPlan=false\n+- HashAggregate(keys=[Student#3414], functions=[finalmerge_avg(merge sum#4487, count#4488L) AS avg(TotScore#3430)#4395])\n   +- Exchange hashpartitioning(Student#3414, 200), ENSURE_REQUIREMENTS, [plan_id=1237]\n      +- HashAggregate(keys=[Student#3414], functions=[partial_avg(TotScore#3430) AS (sum#4487, count#4488L)])\n         +- InMemoryTableScan [Student#3414, TotScore#3430], false\n               +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n                     +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n                        +- *(1) ColumnarToRow\n                           +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 1\navgscore: org.apache.spark.sql.DataFrame = [Student: string, avg(TotScore): double]\n</div>"]}}],"execution_count":0},{"cell_type":"markdown","source":["**Observations**  \nIn the execution plan, we can see that in-memory table scan was used so it signifies that tot_scores were not re-computed and the cache worked properly. \n\nIn the spark jobs, there is shuffling read and write happening since we trigerred an action. Due to the caching some stages has been skipped. This is the desired outcome.\n\n## 2.4 Computing top student analytics\n\nThere are multiple ways to compute top student analytics.\n\nOur roadmap is given below:\n\n1 - we will find the top student by each subject. This brings us the question on whether repartioning should be done or not. Our answer is No: we do not need to check repartioning in this case. Since no dfs from an action were generated and it does not require any further transformations. \n\n2 - we need to determine students with the top score by using groupBy and saving top scores.\n\n\n3 - we will join the top score data frame with the total score data frame based on both the subject and the total score value. This will extract the list of student results that had this top score. \n\n4 - finally, we will print the results."],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"9144b849-abea-4fb7-98cf-ce726dc209b6","inputWidgets":{},"title":""}}},{"cell_type":"code","source":["%scala\n// adding library so we can use sql functions\nimport org.apache.spark.sql.functions._ \n\n// finding top score\nval top = tot_score.groupBy(\"Subject\")\n                   .agg(max(\"TotScore\").alias(\"Tops\"))\n\ntop.show()\n\n// sorting out the students with top scores\nval top_student = tot_score.as(\"t1\")\n                    .join(top.as(\"t2\"),\n                          $\"t2.Tops\" === $\"t1.TotScore\"\n                          && $\"t2.Subject\" === $\"t1.Subject\")\n                    .select(\"t1.Subject\",\"t1.Student\",\"Tops\")\n\ntop_student.show()\n\nprintln(\"--------Explain--------\")\ntop_student.explain\nprintln(\"--------End of Explain--------\")\nprintln(\"# of partitions in dataset : \" + top_student.rdd.getNumPartitions)"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"914059d1-915a-4163-896d-356ca53fdb4b","inputWidgets":{},"title":""}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[{"name":"top","typeStr":"org.apache.spark.sql.DataFrame","schema":{"type":"struct","fields":[{"name":"Subject","type":"string","nullable":true,"metadata":{}},{"name":"Tops","type":"double","nullable":true,"metadata":{}}]},"tableIdentifier":null},{"name":"top_student","typeStr":"org.apache.spark.sql.DataFrame","schema":{"type":"struct","fields":[{"name":"Subject","type":"string","nullable":true,"metadata":{}},{"name":"Student","type":"string","nullable":true,"metadata":{}},{"name":"Tops","type":"double","nullable":true,"metadata":{}}]},"tableIdentifier":null}],"data":"<div class=\"ansiout\">+---------+-------+\n|  Subject|   Tops|\n+---------+-------+\n|     Math|160.175|\n|  Biology|151.725|\n|  Physics|172.475|\n|Chemistry| 120.65|\n+---------+-------+\n\n+---------+-------+-------+\n|  Subject|Student|   Tops|\n+---------+-------+-------+\n|     Math|  James|160.175|\n|     Math|Micheal|160.175|\n|  Biology|  James|151.725|\n|  Physics|   Leny|172.475|\n|Chemistry|   Leny| 120.65|\n+---------+-------+-------+\n\n--------Explain--------\n== Physical Plan ==\nAdaptiveSparkPlan isFinalPlan=false\n+- Project [Subject#3420, Student#3414, Tops#5002]\n   +- BroadcastHashJoin [knownfloatingpointnormalized(normalizenanandzero(TotScore#3430)), Subject#3420], [knownfloatingpointnormalized(normalizenanandzero(Tops#5002)), Subject#5226], Inner, BuildRight, false\n      :- Filter (isnotnull(TotScore#3430) AND isnotnull(Subject#3420))\n      :  +- InMemoryTableScan [Student#3414, Subject#3420, TotScore#3430], [isnotnull(TotScore#3430), isnotnull(Subject#3420)], false\n      :        +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n      :              +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n      :                 +- *(1) ColumnarToRow\n      :                    +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n      +- Exchange SinglePartition, EXECUTOR_BROADCAST, [plan_id=1506]\n         +- Filter isnotnull(Tops#5002)\n            +- HashAggregate(keys=[Subject#5226], functions=[finalmerge_max(merge max#5092) AS max(TotScore#3430)#5001])\n               +- Exchange hashpartitioning(Subject#5226, 200), ENSURE_REQUIREMENTS, [plan_id=1502]\n                  +- HashAggregate(keys=[Subject#5226], functions=[partial_max(TotScore#3430) AS max#5092])\n                     +- Filter isnotnull(Subject#5226)\n                        +- InMemoryTableScan [Subject#5226, TotScore#3430], [isnotnull(Subject#5226)], false\n                              +- InMemoryRelation [Student#5220, ClassScore#5221, TestScore#5222, _c4#5223, _c5#5224, _c6#5225, Subject#5226, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n                                    +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n                                       +- *(1) ColumnarToRow\n                                          +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 4\nimport org.apache.spark.sql.functions._\ntop: org.apache.spark.sql.DataFrame = [Subject: string, Tops: double]\ntop_student: org.apache.spark.sql.DataFrame = [Subject: string, Student: string ... 1 more field]\n</div>","removedWidgets":[],"addedWidgets":{},"metadata":{"isDbfsCommandResult":false},"type":"html","arguments":{}}},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">+---------+-------+\n  Subject|   Tops|\n+---------+-------+\n     Math|160.175|\n  Biology|151.725|\n  Physics|172.475|\nChemistry| 120.65|\n+---------+-------+\n\n+---------+-------+-------+\n  Subject|Student|   Tops|\n+---------+-------+-------+\n     Math|  James|160.175|\n     Math|Micheal|160.175|\n  Biology|  James|151.725|\n  Physics|   Leny|172.475|\nChemistry|   Leny| 120.65|\n+---------+-------+-------+\n\n--------Explain--------\n== Physical Plan ==\nAdaptiveSparkPlan isFinalPlan=false\n+- Project [Subject#3420, Student#3414, Tops#5002]\n   +- BroadcastHashJoin [knownfloatingpointnormalized(normalizenanandzero(TotScore#3430)), Subject#3420], [knownfloatingpointnormalized(normalizenanandzero(Tops#5002)), Subject#5226], Inner, BuildRight, false\n      :- Filter (isnotnull(TotScore#3430) AND isnotnull(Subject#3420))\n      :  +- InMemoryTableScan [Student#3414, Subject#3420, TotScore#3430], [isnotnull(TotScore#3430), isnotnull(Subject#3420)], false\n      :        +- InMemoryRelation [Student#3414, ClassScore#3415, TestScore#3416, _c4#3417, _c5#3418, _c6#3419, Subject#3420, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n      :              +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n      :                 +- *(1) ColumnarToRow\n      :                    +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n      +- Exchange SinglePartition, EXECUTOR_BROADCAST, [plan_id=1506]\n         +- Filter isnotnull(Tops#5002)\n            +- HashAggregate(keys=[Subject#5226], functions=[finalmerge_max(merge max#5092) AS max(TotScore#3430)#5001])\n               +- Exchange hashpartitioning(Subject#5226, 200), ENSURE_REQUIREMENTS, [plan_id=1502]\n                  +- HashAggregate(keys=[Subject#5226], functions=[partial_max(TotScore#3430) AS max#5092])\n                     +- Filter isnotnull(Subject#5226)\n                        +- InMemoryTableScan [Subject#5226, TotScore#3430], [isnotnull(Subject#5226)], false\n                              +- InMemoryRelation [Student#5220, ClassScore#5221, TestScore#5222, _c4#5223, _c5#5224, _c6#5225, Subject#5226, TotScore#3430], StorageLevel(disk, memory, deserialized, 1 replicas)\n                                    +- *(1) Project [Student#127, ClassScore#128, TestScore#129, _c4#130, _c5#131, _c6#132, Subject#133, (cast(ClassScore#128 as double) + TestScore#129) AS TotScore#356]\n                                       +- *(1) ColumnarToRow\n                                          +- FileScan parquet [Student#127,ClassScore#128,TestScore#129,_c4#130,_c5#131,_c6#132,Subject#133] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[dbfs:/FileStore/FileStore/partitioned_st], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;Student:string,ClassScore:int,TestScore:double,_c4:string,_c5:string,_c6:string&gt;\n\n\n--------End of Explain--------\n# of partitions in dataset : 4\nimport org.apache.spark.sql.functions._\ntop: org.apache.spark.sql.DataFrame = [Subject: string, Tops: double]\ntop_student: org.apache.spark.sql.DataFrame = [Subject: string, Student: string ... 1 more field]\n</div>"]}}],"execution_count":0},{"cell_type":"markdown","source":["**Observations**  \nStudents with top scores are printed correctly and there are two students with equal score on math.   \nIn the execution plan, the cache is employed.  \nSpark Optimizer resulted in a **broadcast hash join** for broadcasting top score df. \nIt also trigerred re-execution of aggregation. This part can be imporoved by adding another caching."],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"6206ada1-c3c5-468e-95ad-6615c07df929","inputWidgets":{},"title":""}}},{"cell_type":"markdown","source":["# 3- Conclusions\n\n\nIn this notebook, we started with a CSV file including student scores by subject for a school year. It had four attributes: student name, subject, class score, and test score.\n\nFirst, we loaded data into HDFS in Parquet format with GZIP compression (a partitioning scheme) and then read the data. C\nThen, we computed the total score for each student per subject, printed physics scores, computed average total score across all subjects, and found the highest scoring student for each subject. \n\nAdditionally, further analytics can be done on data by including sql function libraries, applying ML models and graphing.\n\nFor the mechanics of SQL aggregation with Spark, one needs to include library below: \n\n*import org.apache.spark.sql.functions._*\n* GroupedData\n    * `.mean()`\n    * `.sum()`\n    * `.count()`\n    * Other aggregations\n    * `.agg(exprs)`\n        * exprs as dict\n        * exprs as list\n        * advanced expressions"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"ec1887b7-f22a-4ce1-b026-08c5b5d2683e","inputWidgets":{},"title":""}}},{"cell_type":"markdown","source":["End of Notebook\n# --- END ---"],"metadata":{"application/vnd.databricks.v1+cell":{"showTitle":false,"cellMetadata":{},"nuid":"5e291b58-3ac5-4bd6-9bd0-aecc8eddd06a","inputWidgets":{},"title":""}}}],"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"mimetype":"text/x-python","name":"python","pygments_lexer":"ipython3","codemirror_mode":{"name":"ipython","version":3},"version":"3.8.3","nbconvert_exporter":"python","file_extension":".py"},"application/vnd.databricks.v1+notebook":{"notebookName":"L01-Project_Big Data Analytics with Hadoop and Apache Spark","dashboards":[],"notebookMetadata":{"pythonIndentUnit":2},"language":"python","widgets":{},"notebookOrigID":3887246253406713}},"nbformat":4,"nbformat_minor":0}