<jupyter_start><jupyter_text>Exercise 1How many distinct products have been sold in each store?<jupyter_code>df = spark.read.format("csv").option("header", "true").load("../data/sales.csv")
df.printSchema()
df.select("store_id", "product_id").distinct().show()<jupyter_output>+--------+----------+
|store_id|product_id|
+--------+----------+
|       2|        22|
|       2|       138|
|       2|      1245|
|       2|       756|
|       3|       502|
|       3|       310|
|       3|      1009|
|       3|       315|
|       3|        42|
|       3|      1486|
|       3|        49|
|       3|       587|
|       3|       539|
|       6|      1238|
|       6|      1069|
|       6|      1401|
|       6|       152|
|       6|       162|
|       7|       273|
|       7|       752|
+--------+----------+
only showing top 20 rows<jupyter_text>Exercise 2<jupyter_code>df = spark.read.format("csv").option("header", "true").load("../data/online-retail-dataset.csv")
df.printSchema()<jupyter_output>root
 |-- InvoiceNo: string (nullable = true)
 |-- StockCode: string (nullable = true)
 |-- Description: string (nullable = true)
 |-- Quantity: string (nullable = true)
 |-- InvoiceDate: string (nullable = true)
 |-- UnitPrice: string (nullable = true)
 |-- CustomerID: string (nullable = true)
 |-- Country: string (nullable = true)<jupyter_text>**Exercise 2.1**What is the average quantity bought by the customer 14769?<jupyter_code>from pyspark.sql.functions import col, avg
df.where(df.CustomerID == "14769").agg(avg(col("Quantity"))).collect()[0][0]<jupyter_output><empty_output><jupyter_text>**Exercise 2.2**What is the most occurring word in the items bought by the customers from France?<jupyter_code>from pyspark.sql.functions import explode, split, concat, lit, desc
words = df.where(df.Country == "France").select("Description").select(explode(split(col("Description"), " ")).alias("word"))

words_count = words.groupby(words.word).count().orderBy(col("count").desc())

words_count.collect()[1][0]<jupyter_output><empty_output><jupyter_text>Exercise 3<jupyter_code>data_rdd = spark.sparkContext.textFile("../data/employee_info.csv")<jupyter_output><empty_output><jupyter_text>**Exercise 3.1**How many police officers are there at the POLICE department?<jupyter_code>def filter_fun(x):
    return True if "POLICE OFFICER" in x else False
data_rdd.filter(filter_fun).count()<jupyter_output><empty_output><jupyter_text>**Exercise 3.2**Who is the lieutenant with the highest salary at the FIRE department?<jupyter_code>import re

from operator import add
def filter_fun(x):
    return True if "FIRE" in x and re.search(r"\b,LIEUTENANT,\b", x) is not None else False

def map_fun(x):
    values = x.split(",")
    
    return (float(values[-2]), values[1] + " " + values[0])
fire_lieutenant_rdd = data_rdd.filter(filter_fun)

fire_lieutenant_rdd.map(map_fun).sortByKey(ascending=False).take(1) 
# Stop the spark context
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17071524966261808, 'fraction_numerical': 0.038461538461538464, 'mean_word_length': 3.379615952732644, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '274487', 'license': 'MIT', 'n_tokens_mistral': 1081, 'n_tokens_neox': 1026, 'n_words': 215, 'path': 'labs/lab-5/notebooks/lab5_exercises.ipynb', 'repo_name': 'tomdewildt/data-engineering', 'size': '2964'}