Retail_Data_Analytics.ipynb

{"cells":[{"cell_type":"code","source":["from pyspark.sql.types import StructField, StructType, StringType, LongType, DoubleType, TimestampType\n\n# Task 0 Load the “/databricks-datasets/definitive-guide/data/retail-data/all/*.csv” file into a dataframe.\n\nmySchema = StructType([\n    StructField(\"InvoiceNo\",LongType(),True),\n    StructField(\"StockCode\",StringType(),True),\n    StructField(\"Description\",StringType(),True),\n    StructField(\"Quantity\",StringType(),True),\n    StructField(\"InvoiceDate\",TimestampType(),True),\n    StructField(\"UnitPrice\",DoubleType(),True),\n    StructField(\"CustomerID\",LongType(),True),\n    StructField(\"Country\",StringType(),True)])\n\ndf = spark.read.format(\"csv\").schema(mySchema).load(\"/databricks-datasets/definitive-guide/data/retail-data/all/*.csv\", header='true', timestampFormat = \"M/d/y h:m\")\ndf.show(5)"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"eb22654a-8e4c-47c5-8395-fcb2469bb372"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+\n|InvoiceNo|StockCode|         Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|\n+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+\n|   536365|   85123A|WHITE HANGING HEA...|       6|2010-12-01 08:26:00|     2.55|     17850|United Kingdom|\n|   536365|    71053| WHITE METAL LANTERN|       6|2010-12-01 08:26:00|     3.39|     17850|United Kingdom|\n|   536365|   84406B|CREAM CUPID HEART...|       8|2010-12-01 08:26:00|     2.75|     17850|United Kingdom|\n|   536365|   84029G|KNITTED UNION FLA...|       6|2010-12-01 08:26:00|     3.39|     17850|United Kingdom|\n|   536365|   84029E|RED WOOLLY HOTTIE...|       6|2010-12-01 08:26:00|     3.39|     17850|United Kingdom|\n+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+\nonly showing top 5 rows\n\n","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"ansi","arguments":{}}},"output_type":"display_data","data":{"text/plain":["+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+\n|InvoiceNo|StockCode|         Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|\n+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+\n|   536365|   85123A|WHITE HANGING HEA...|       6|2010-12-01 08:26:00|     2.55|     17850|United Kingdom|\n|   536365|    71053| WHITE METAL LANTERN|       6|2010-12-01 08:26:00|     3.39|     17850|United Kingdom|\n|   536365|   84406B|CREAM CUPID HEART...|       8|2010-12-01 08:26:00|     2.75|     17850|United Kingdom|\n|   536365|   84029G|KNITTED UNION FLA...|       6|2010-12-01 08:26:00|     3.39|     17850|United Kingdom|\n|   536365|   84029E|RED WOOLLY HOTTIE...|       6|2010-12-01 08:26:00|     3.39|     17850|United Kingdom|\n+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+\nonly showing top 5 rows\n\n"]},"transient":null}],"execution_count":0},{"cell_type":"code","source":["# Task 1 (20 pts). Count the number of items/StockCode for each InvoiceNo.\n\nfrom pyspark.sql.functions import col, countDistinct\n\n# Using countDistinct so that one type of item is counted only once\n\ndf.groupBy(col(\"InvoiceNo\")).agg(countDistinct(col(\"StockCode\")).alias(\"Number_of_unique_items\")).show(5)"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"40e7ccce-233b-4505-a23d-b2eb914be9a0"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"+---------+----------------------+\n|InvoiceNo|Number_of_unique_items|\n+---------+----------------------+\n|   540168|                    51|\n|   540260|                    41|\n|   538809|                    11|\n|   541864|                    19|\n|   537381|                    24|\n+---------+----------------------+\nonly showing top 5 rows\n\n","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"ansi","arguments":{}}},"output_type":"display_data","data":{"text/plain":["+---------+----------------------+\n|InvoiceNo|Number_of_unique_items|\n+---------+----------------------+\n|   540168|                    51|\n|   540260|                    41|\n|   538809|                    11|\n|   541864|                    19|\n|   537381|                    24|\n+---------+----------------------+\nonly showing top 5 rows\n\n"]},"transient":null}],"execution_count":0},{"cell_type":"code","source":["# Task 2 (20 pts). For each customer, calculate the total amount of spendings.\n\nfrom pyspark.sql.functions import col, sum\n\n# Generate spending for single row = unit price * qty\n# Group by the customer id and take sum of all the spendings\n\ndf.withColumn(\"Spending\",col(\"UnitPrice\") * col(\"Quantity\")).groupBy(col(\"CustomerID\")).agg(sum(col(\"Spending\")).alias(\"Total_spendings\")).show(5)"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"3c37c248-ecc2-415d-a24b-9fc09b2aa712"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"+----------+------------------+\n|CustomerID|   Total_spendings|\n+----------+------------------+\n|     15194| 7521.170000000001|\n|     17048|            864.32|\n|     13098|28658.879999999997|\n|     16781|            294.65|\n|     12967|           1194.75|\n+----------+------------------+\nonly showing top 5 rows\n\n","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"ansi","arguments":{}}},"output_type":"display_data","data":{"text/plain":["+----------+------------------+\n|CustomerID|   Total_spendings|\n+----------+------------------+\n|     15194| 7521.170000000001|\n|     17048|            864.32|\n|     13098|28658.879999999997|\n|     16781|            294.65|\n|     12967|           1194.75|\n+----------+------------------+\nonly showing top 5 rows\n\n"]},"transient":null}],"execution_count":0},{"cell_type":"code","source":["# Task 3 (20 pts). For each item/StockCode, list the total amount of sold products.\n\n# Get single amount = unit price * qty\n# Group by stockcode and get total of all the amounts\n\ndf.withColumn(\"Amount\",col(\"UnitPrice\") * col(\"Quantity\")).groupBy(col(\"StockCode\")).agg(sum(col(\"Amount\")).alias(\"Total_amount\")).show(5)"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"5b0b29e8-054c-416e-9b92-7e56b73e412b"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"+---------+------------------+\n|StockCode|      Total_amount|\n+---------+------------------+\n|    22728|20548.399999999998|\n|    21889| 8829.790000000003|\n|   90210B|             61.56|\n|    21259| 7014.750000000002|\n|    21894| 908.3299999999999|\n+---------+------------------+\nonly showing top 5 rows\n\n","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"ansi","arguments":{}}},"output_type":"display_data","data":{"text/plain":["+---------+------------------+\n|StockCode|      Total_amount|\n+---------+------------------+\n|    22728|20548.399999999998|\n|    21889| 8829.790000000003|\n|   90210B|             61.56|\n|    21259| 7014.750000000002|\n|    21894| 908.3299999999999|\n+---------+------------------+\nonly showing top 5 rows\n\n"]},"transient":null}],"execution_count":0},{"cell_type":"code","source":["# Task 4 (20 pts). For each InvoiceDate (Attention: not timestamp), calculate the difference of total sales amount with the previous date. \nspark.sql(\"set spark.sql.legacy.timeParserPolicy=LEGACY\")\nfrom pyspark.sql.functions import window, lag\nfrom pyspark.sql.window import Window\n\n# Get single Sale value = unit price * qty for each row. \n# Group by invoice date (using \"window\" to get values for 1 day) and calculate sum of the sales\n# This gives columns INVOICE DATE and TOTAL SALE (for that date)\n\nnew_df = df.withColumn(\"Sale\", col(\"UnitPrice\") * col(\"Quantity\")).groupBy(window(col(\"InvoiceDate\"), \"1 day\")).agg(sum(col(\"Sale\"))).withColumn('Invoice_date',col('window')['start'].cast('date')).select(\"Invoice_date\", col(\"sum(Sale)\").alias(\"Total_sale\"))\n\n# Order by INVOICE DATE to get previous value and use lag to get previous value for TOTAL SALE\n\nlagCol = lag(col(\"Total_sale\"), 1).over(Window.orderBy(\"Invoice_date\"))\n\n# Get SALE_DIFFERENCE = current value - previous day value\n# Show columns INVOICE DATE, TOTAL SALE (for that date), PREVIOUS DATE SALE and SALE_DIFFERENCE\n\nnew_df.withColumn(\"Prev_day_sale\", lagCol).na.fill(0).withColumn(\"Sale_difference\", col(\"Total_sale\") - col(\"Prev_day_sale\")).show(5)\n"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"1c1f19bb-4a2c-4de1-a4ac-7c8630cae327"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"+------------+------------------+-----------------+-------------------+\n|Invoice_date|        Total_sale|    Prev_day_sale|    Sale_difference|\n+------------+------------------+-----------------+-------------------+\n|  2010-12-01| 58635.56000000026|              0.0|  58635.56000000026|\n|  2010-12-02| 46207.27999999991|58635.56000000026|-12428.280000000348|\n|  2010-12-03|  45620.4599999999|46207.27999999991| -586.8200000000143|\n|  2010-12-05| 31383.95000000016| 45620.4599999999|-14236.509999999736|\n|  2010-12-06|53860.180000000015|31383.95000000016| 22476.229999999854|\n+------------+------------------+-----------------+-------------------+\nonly showing top 5 rows\n\n","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"ansi","arguments":{}}},"output_type":"display_data","data":{"text/plain":["+------------+------------------+-----------------+-------------------+\n|Invoice_date|        Total_sale|    Prev_day_sale|    Sale_difference|\n+------------+------------------+-----------------+-------------------+\n|  2010-12-01| 58635.56000000026|              0.0|  58635.56000000026|\n|  2010-12-02| 46207.27999999991|58635.56000000026|-12428.280000000348|\n|  2010-12-03|  45620.4599999999|46207.27999999991| -586.8200000000143|\n|  2010-12-05| 31383.95000000016| 45620.4599999999|-14236.509999999736|\n|  2010-12-06|53860.180000000015|31383.95000000016| 22476.229999999854|\n+------------+------------------+-----------------+-------------------+\nonly showing top 5 rows\n\n"]},"transient":null}],"execution_count":0},{"cell_type":"code","source":["# Task 5 (20 pts). For each InvoiceDate (Attention: not timestamp), calculate the difference of the purchase countries with the next date. \n\nfrom pyspark.sql.functions import collect_set, lead, array_except\n\n# Group by invoice date (using \"window\" to get values for 1 day) and get set of countries (collect_set removes duplicates)\n# This gives columns INVOICE DATE and COUNTRIES (array)\n\nnew_df = df.groupBy(window(col(\"InvoiceDate\"), \"1 day\")).agg(collect_set(col(\"Country\"))).withColumn('Invoice_date',col('window')['start'].cast('date')).select(\"Invoice_date\", col(\"collect_set(Country)\").alias(\"Countries\"))\n\n# Order by INVOICE DATE to get next value and use lead to get next value for COUNTRIES\n\nleadCol = lead(col(\"Countries\"), 1).over(Window.orderBy(\"Invoice_date\"))\n\n# Get Country difference = {Next day countries} - {Countries} using array_except\n# Show columns INVOICE DATE, Countries, Next_day_countries, Country_difference\n\nnew_df.withColumn(\"Next_date_countries\", leadCol).withColumn(\"Country_difference\", array_except(col(\"Countries\"),col(\"Next_date_countries\"))).show(5)"],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"fb5bd71c-5d02-4c8b-95f5-7f558f4a9376"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"datasetInfos":[],"data":"+------------+--------------------+--------------------+--------------------+\n|Invoice_date|           Countries| Next_date_countries|  Country_difference|\n+------------+--------------------+--------------------+--------------------+\n|  2010-12-01|[France, Australi...|[EIRE, Germany, U...|[France, Australi...|\n|  2010-12-02|[EIRE, Germany, U...|[France, Portugal...|                  []|\n|  2010-12-03|[France, Portugal...|[France, Germany,...|[Portugal, Italy,...|\n|  2010-12-05|[France, Germany,...|[Portugal, Italy,...|[France, Japan, L...|\n|  2010-12-06|[Portugal, Italy,...|[France, Germany,...|[Portugal, Italy,...|\n+------------+--------------------+--------------------+--------------------+\nonly showing top 5 rows\n\n","removedWidgets":[],"addedWidgets":{},"metadata":{},"type":"ansi","arguments":{}}},"output_type":"display_data","data":{"text/plain":["+------------+--------------------+--------------------+--------------------+\n|Invoice_date|           Countries| Next_date_countries|  Country_difference|\n+------------+--------------------+--------------------+--------------------+\n|  2010-12-01|[France, Australi...|[EIRE, Germany, U...|[France, Australi...|\n|  2010-12-02|[EIRE, Germany, U...|[France, Portugal...|                  []|\n|  2010-12-03|[France, Portugal...|[France, Germany,...|[Portugal, Italy,...|\n|  2010-12-05|[France, Germany,...|[Portugal, Italy,...|[France, Japan, L...|\n|  2010-12-06|[Portugal, Italy,...|[France, Germany,...|[Portugal, Italy,...|\n+------------+--------------------+--------------------+--------------------+\nonly showing top 5 rows\n\n"]},"transient":null}],"execution_count":0},{"cell_type":"code","source":[""],"metadata":{"application/vnd.databricks.v1+cell":{"title":"","showTitle":false,"inputWidgets":{},"nuid":"413a81d8-8165-4d17-b673-f5e2300bdf84"}},"outputs":[{"output_type":"display_data","metadata":{"application/vnd.databricks.v1+output":{"data":"","errorSummary":"","metadata":{},"errorTraceType":null,"type":"ipynbError","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>"]},"transient":null}],"execution_count":0}],"metadata":{"application/vnd.databricks.v1+notebook":{"notebookName":"Assignment_2","dashboards":[],"notebookMetadata":{"pythonIndentUnit":4},"language":"python","widgets":{},"notebookOrigID":2002959216030285}},"nbformat":4,"nbformat_minor":0}