{
    "metadata": {
        "kernelspec": {
            "name": "pyspark3kernel",
            "display_name": "PySpark3"
        },
        "language_info": {
            "name": "pyspark3",
            "mimetype": "text/x-python",
            "codemirror_mode": {
                "name": "python",
                "version": 3
            },
            "pygments_lexer": "python3"
        }
    },
    "nbformat_minor": 2,
    "nbformat": 4,
    "cells": [
        {
            "cell_type": "markdown",
            "source": "# Spark sample showing read/write methods\nIn this sample notebook, we will read CSV file from HDFS, write it as parquet file and save a Hive table definition. We will also run some Spark SQL commands using the Hive table.\n",
            "metadata": {}
        },
        {
            "cell_type": "code",
            "source": "# Read the clickstream CSV file(s) into a spark data frame, print schema & top rows\nresults = spark.read.option(\"inferSchema\", \"true\").csv('/clickstream_data').toDF(\n            \"wcs_click_date_sk\", \"wcs_click_time_sk\", \"wcs_sales_sk\", \"wcs_item_sk\", \"wcs_web_page_sk\", \"wcs_user_sk\"\n            )\nresults.printSchema()\nresults.show()",
            "metadata": {},
            "outputs": [
                {
                    "name": "stdout",
                    "text": "root\n |-- wcs_click_date_sk: integer (nullable = true)\n |-- wcs_click_time_sk: integer (nullable = true)\n |-- wcs_sales_sk: integer (nullable = true)\n |-- wcs_item_sk: integer (nullable = true)\n |-- wcs_web_page_sk: integer (nullable = true)\n |-- wcs_user_sk: integer (nullable = true)\n\n+-----------------+-----------------+------------+-----------+---------------+-----------+\n|wcs_click_date_sk|wcs_click_time_sk|wcs_sales_sk|wcs_item_sk|wcs_web_page_sk|wcs_user_sk|\n+-----------------+-----------------+------------+-----------+---------------+-----------+\n|            36890|            40052|        null|       4379|             34|       null|\n|            36890|            41285|        null|       6245|             34|       null|\n|            36890|            23115|        null|      13852|             34|       null|\n|            36890|            17702|        null|      15975|             34|       null|\n|            36890|            62676|        null|       2119|             34|       null|\n|            36890|            34267|        null|      10273|             34|       null|\n|            36890|             8502|        null|      17790|             34|       null|\n|            36890|            54340|        null|       3453|             34|       null|\n|            36890|            54370|        null|       6372|             34|       null|\n|            36890|             6578|        null|      17203|             34|       null|\n|            36890|            75088|        null|       4891|             34|       null|\n|            36890|            23922|        null|      11332|             34|       null|\n|            36890|            28761|        null|       4484|             34|       null|\n|            36890|            21444|        null|       5582|             34|       null|\n|            36890|            58917|        null|       8833|             34|       null|\n|            36890|            27578|        null|       8599|             34|       null|\n|            36890|             8059|        null|       6720|             34|       null|\n|            36890|            43008|        null|      17175|             34|       null|\n|            36890|             4378|        null|      10644|             34|       null|\n|            36890|            55403|        null|       8139|             34|       null|\n+-----------------+-----------------+------------+-----------+---------------+-----------+\nonly showing top 20 rows",
                    "output_type": "stream"
                }
            ],
            "execution_count": 3
        },
        {
            "cell_type": "code",
            "source": "# Disable saving SUCCESS file\r\nsc._jsc.hadoopConfiguration().set(\"mapreduce.fileoutputcommitter.marksuccessfuljobs\", \"false\") \r\n\r\n# Print the current warehouse directory where the parquet files will be stored\r\nprint(spark.conf.get(\"spark.sql.warehouse.dir\"))\r\n\r\n# Save results as parquet & orc file and create hive table\r\nresults.write.format(\"parquet\").mode(\"overwrite\").saveAsTable(\"web_clickstreams\")\r\nresults.write.format(\"orc\").mode(\"overwrite\").saveAsTable(\"web_clickstreams_orc\")",
            "metadata": {},
            "outputs": [
                {
                    "name": "stdout",
                    "text": "hdfs:///user/hive/warehouse",
                    "output_type": "stream"
                }
            ],
            "execution_count": 4
        },
        {
            "cell_type": "code",
            "source": "# Execute Spark SQL commands\r\nsqlDF = spark.sql(\"SELECT * FROM web_clickstreams LIMIT 100\")\r\nsqlDF.show()\r\n\r\nsqlDF = spark.sql(\"SELECT wcs_user_sk, COUNT(*)\\\r\n                     FROM web_clickstreams\\\r\n                    WHERE wcs_user_sk IS NOT NULL\\\r\n                   GROUP BY wcs_user_sk\\\r\n                   ORDER BY COUNT(*) DESC LIMIT 100\")\r\nsqlDF.show()",
            "metadata": {},
            "outputs": [
                {
                    "name": "stdout",
                    "text": "+-----------------+-----------------+------------+-----------+---------------+-----------+\n|wcs_click_date_sk|wcs_click_time_sk|wcs_sales_sk|wcs_item_sk|wcs_web_page_sk|wcs_user_sk|\n+-----------------+-----------------+------------+-----------+---------------+-----------+\n|            37506|             7933|        null|       1384|              2|      39437|\n|            37506|            56044|        null|      14689|              2|      26419|\n|            37506|            52706|        null|       8541|              2|      44016|\n|            37506|            67325|        null|      16129|              2|      83371|\n|            37506|            84857|        null|       1869|              2|      13090|\n|            37506|            49599|        null|       2994|              2|       8940|\n|            37506|            78150|        null|      11392|              2|      65633|\n|            37506|            38720|        null|      14366|              2|      22281|\n|            37506|            79915|        null|      11102|              2|      81755|\n|            37506|            67253|        null|       5380|              2|      46868|\n|            37506|             6507|        null|       6813|              2|      49363|\n|            37506|            18280|        null|       1458|              2|      49363|\n|            37506|            72258|        null|       2869|              2|      67756|\n|            37506|             8045|        null|        615|              2|      86035|\n|            37506|            86164|        null|       7000|              2|      94821|\n|            37506|            29724|        null|       2767|              2|      94821|\n|            37506|            55471|        null|       3584|              2|      62792|\n|            37506|              677|        null|       1720|              2|      27212|\n|            37506|            66638|        null|       9898|              2|      20370|\n|            37506|            48515|        null|       9394|              2|      17157|\n+-----------------+-----------------+------------+-----------+---------------+-----------+\nonly showing top 20 rows\n\n+-----------+--------+\n|wcs_user_sk|count(1)|\n+-----------+--------+\n|      65042|     832|\n|      55928|     821|\n|      15570|     791|\n|      31138|     788|\n|      68188|     784|\n|      88205|     760|\n|      15678|     757|\n|      48063|     741|\n|      77518|     741|\n|      92978|     728|\n|      82129|     727|\n|      21700|     725|\n|      69707|     724|\n|      38895|     719|\n|      97643|     716|\n|      74426|     707|\n|       7813|     704|\n|      49528|     700|\n|      55766|     698|\n|      54355|     697|\n+-----------+--------+\nonly showing top 20 rows",
                    "output_type": "stream"
                }
            ],
            "execution_count": 5
        },
        {
            "cell_type": "code",
            "source": "# Read the product reviews CSV files into a spark data frame, print schema & top rows\r\nresults = spark.read.option(\"inferSchema\", \"true\").csv('/product_review_data').toDF(\r\n            \"pr_review_sk\", \"pr_review_content\"\r\n            )\r\nresults.printSchema()\r\nresults.show()",
            "metadata": {},
            "outputs": [
                {
                    "name": "stdout",
                    "text": "root\n |-- pr_review_sk: integer (nullable = true)\n |-- pr_review_content: string (nullable = true)\n\n+------------+--------------------+\n|pr_review_sk|   pr_review_content|\n+------------+--------------------+\n|       72621|Works fine. Easy ...|\n|       89334|great product to ...|\n|       89335|Next time will go...|\n|       84259|Great Gift Great ...|\n|       84398|After trip to Par...|\n|       66434|Simply the best t...|\n|       66501|This is the exact...|\n|       66587|Not super magnet;...|\n|       66680|Installed as bath...|\n|       66694|Our home was buil...|\n|       84489|Hi ;We are runnin...|\n|       79052|Terra cotta is th...|\n|       73034|One of my fingern...|\n|       73298|We installed thes...|\n|       66810|needed silicone c...|\n|       66912|Great Gift Great ...|\n|       67028|Laguiole knives a...|\n|       89770|Good sound timers...|\n|       84679|AWESOME FEEDBACK ...|\n|       84953|love the retro gl...|\n+------------+--------------------+\nonly showing top 20 rows",
                    "output_type": "stream"
                }
            ],
            "execution_count": 6
        },
        {
            "cell_type": "code",
            "source": "# Save results as parquet, and orc formats and create hive table\r\nresults.write.format(\"parquet\").mode(\"overwrite\").saveAsTable(\"product_reviews\")\r\nresults.write.format(\"orc\").mode(\"overwrite\").saveAsTable(\"product_reviews_orc\")\r\n",
            "metadata": {},
            "outputs": [],
            "execution_count": 7
        },
        {
            "cell_type": "code",
            "source": "# Execute Spark SQL commands\r\nsqlDF = spark.sql(\"SELECT pr_review_sk, CHAR_LENGTH(pr_review_content) as len FROM product_reviews LIMIT 100\")\r\nsqlDF.show()",
            "metadata": {},
            "outputs": [
                {
                    "name": "stdout",
                    "text": "+------------+----+\n|pr_review_sk| len|\n+------------+----+\n|       26035| 876|\n|       26037| 109|\n|       26038| 478|\n|       26041| 106|\n|       26043| 332|\n|       26044| 487|\n|       26045| 428|\n|       26048|  87|\n|       26049| 118|\n|       26051|2906|\n|       26053| 464|\n|       26054| 212|\n|       26059| 191|\n|       26060| 207|\n|       26061| 515|\n|       26063|  59|\n|       26069| 487|\n|       26070| 160|\n|       26071| 380|\n|       26072| 234|\n+------------+----+\nonly showing top 20 rows",
                    "output_type": "stream"
                }
            ],
            "execution_count": 8
        }
    ]
}}