<jupyter_start><jupyter_text>Imports<jupyter_code>from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType
spark = (SparkSession.builder.appName("SalesAnalytics").getOrCreate())<jupyter_output><empty_output><jupyter_text>Data Preparation<jupyter_code>schema = StructType([
    StructField("Order ID", StringType(), True),
    StructField("Product", StringType(), True),
    StructField("Quantity Ordered", StringType(), True),
    StructField("Price Each", StringType(), True),
    StructField("Order Date", StringType(), True),
    StructField("Purchase Address", StringType(), True)   
])
sales_data_fpath = "./data/salesdata"
sales_raw_df = (spark.read.format("csv")
                .option("header",True)
                .schema(schema)
                .load(sales_data_fpath))
sales_raw_df.show(10)
sales_raw_df.printSchema()<jupyter_output>root
 |-- Order ID: string (nullable = true)
 |-- Product: string (nullable = true)
 |-- Quantity Ordered: string (nullable = true)
 |-- Price Each: string (nullable = true)
 |-- Order Date: string (nullable = true)
 |-- Purchase Address: string (nullable = true)<jupyter_text>Data Preparation and Cleansing  Remove Null Rows and Bad Records<jupyter_code>from pyspark.sql.functions import col
## show null values in dataframe
sales_raw_df.filter(col("Order ID").isNull() == True).show(10)
sales_raw_df = sales_raw_df.na.drop("any")
## No more Null Values
sales_raw_df.filter(col("Order ID").isNull() == True).show(10)
sales_raw_df.describe().show()
sales_raw_df.filter(col("Order ID") == "Order ID").show(10)
sales_temp_df = sales_raw_df.distinct()
sales_temp_df.filter(col("Order ID") == "Order ID").show(10)
sales_temp_df = sales_temp_df.filter(col("Order ID") != "Order ID")
sales_temp_df.filter(col("Order ID") == "Order ID").show(10)
sales_temp_df.show(10, truncate=False)
sales_temp_df.describe().show()<jupyter_output>+-------+------------------+------------+------------------+------------------+--------------+--------------------+
|summary|          Order ID|     Product|  Quantity Ordered|        Price Each|    Order Date|    Purchase Address|
+-------+------------------+------------+------------------+------------------+--------------+--------------------+
|  count|            185686|      185686|            185686|            185686|        185686|              185686|
|   mean|230411.37622653297|        null|1.1245435843305365|184.51925546352427|          null|                null|
| stddev| 51511.71718332086|        null|0.4430687383832874| 332.8438383900525|          null|                null|
|    min|            141234|20in Monitor|                 1|            109.99|01/01/19 03:07|1 11th St, Atlant...|
|    max|            319670|      iPhone|                 9|            999.99|12/31/19 23:53|999 Wilson St, Sa...|
+-------+------------------+------------+------------------+------------[...]<jupyter_text>Extract the city and State from Purchase Address<jupyter_code>from pyspark.sql.functions import split
sales_temp_df.select("Purchase Address").show(10, False)
sales_temp_df.select("Purchase Address" ,split(col("purchase Address"), ",")).show(10, False)
## Get City
sales_temp_df.select("Purchase Address" ,split(col("purchase Address"), ",").getItem(1)).show(10, False)
## Get State
sales_temp_df.select("Purchase Address" ,split(col("purchase Address"), ",").getItem(2)).show(10, False)
## Get State--- We use an Extra Split function to drill into the first split
sales_temp_df.select("Purchase Address" ,split(split(col("purchase Address"), ",").getItem(2), ' ')).show(10, False)
sales_temp_df = (sales_temp_df.withColumn("City", split(col("purchase Address"), ",").getItem(1))
                 .withColumn("State", split(split(col("purchase Address"), ",").getItem(2), ' ').getItem(1)))
sales_temp_df.show(10, False)<jupyter_output>+--------+--------------------------+----------------+----------+--------------+----------------------------------------+--------------+-----+
|Order ID|Product                   |Quantity Ordered|Price Each|Order Date    |Purchase Address                        |City          |State|
+--------+--------------------------+----------------+----------+--------------+----------------------------------------+--------------+-----+
|295900  |AA Batteries (4-pack)     |1               |3.84      |12/27/19 18:56|283 Washington St, Boston, MA 02215     | Boston       |MA   |
|295923  |Lightning Charging Cable  |1               |14.95     |12/21/19 13:41|968 8th St, Austin, TX 73301            | Austin       |TX   |
|295991  |Lightning Charging Cable  |1               |14.95     |12/15/19 20:16|857 Center St, Boston, MA 02215         | Boston       |MA   |
|296076  |Macbook Pro Laptop        |1               |1700      |12/03/19 15:19|679 Chestnut St, San Francisco, CA 94016| San Francisco|CA   |[...]<jupyter_text>Rename and Change DataTypes<jupyter_code>from pyspark.sql.functions import to_timestamp , year , month
from pyspark.sql.types import IntegerType , FloatType
sales_temp_df = (sales_temp_df.withColumn("OrderID", col("Order ID").cast(IntegerType()))
                              .withColumn("Quantity", col("Quantity Ordered").cast(IntegerType()))
                              .withColumn("Price", col("Price Each").cast(FloatType()))
                              .withColumn("OrderDate", to_timestamp(col("Order Date"), "MM/dd/yy HH:mm"))
                              .withColumnRenamed("Purchase Address", "StoreAddress")
                              .drop("order ID")
                              .drop("Quantity Ordered")
                              .drop("Price Each")
                              .drop("Purchase Address"))
sales_temp_df.show()
sales_temp_df.printSchema()<jupyter_output>root
 |-- Product: string (nullable = true)
 |-- Order Date: string (nullable = true)
 |-- StoreAddress: string (nullable = true)
 |-- City: string (nullable = true)
 |-- State: string (nullable = true)
 |-- OrderID: integer (nullable = true)
 |-- Quantity: integer (nullable = true)
 |-- Price: float (nullable = true)
 |-- OrderDate: timestamp (nullable = true)<jupyter_text>Add New Columns:Month and Year<jupyter_code>sales_temp_df = (sales_temp_df.withColumn("ReportYear", year(col("OrderDate")))
                              .withColumn("Month", month(col("OrderDate"))))
sales_temp_df.show(10)<jupyter_output>+--------------------+--------------+--------------------+--------------+-----+-------+--------+------+-------------------+----------+-----+
|             Product|    Order Date|        StoreAddress|          City|State|OrderID|Quantity| Price|          OrderDate|ReportYear|Month|
+--------------------+--------------+--------------------+--------------+-----+-------+--------+------+-------------------+----------+-----+
|AA Batteries (4-p...|12/27/19 18:56|283 Washington St...|        Boston|   MA| 295900|       1|  3.84|2019-12-27 18:56:00|      2019|   12|
|Lightning Chargin...|12/21/19 13:41|968 8th St, Austi...|        Austin|   TX| 295923|       1| 14.95|2019-12-21 13:41:00|      2019|   12|
|Lightning Chargin...|12/15/19 20:16|857 Center St, Bo...|        Boston|   MA| 295991|       1| 14.95|2019-12-15 20:16:00|      2019|   12|
|  Macbook Pro Laptop|12/03/19 15:19|679 Chestnut St, ...| San Francisco|   CA| 296076|       1|1700.0|2019-12-03 15:19:00|      2019|   12|
|AAA Batterie[...]
{'fraction_non_alphanumeric': 0.2451760895965457, 'fraction_numerical': 0.06976116583457023, 'mean_word_length': 1.9983818770226538, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '64413', 'license': 'MIT', 'n_tokens_mistral': 2545, 'n_tokens_neox': 2165, 'n_words': 467, 'path': 'sparkdf/SparkSalesAnalytics.ipynb', 'repo_name': 'Benniah/PySpark', 'size': '7411'}