Big Data Analysis with PySpark

# From the project root (big-data-analysis-with-mogodb)
python3 -m venv .venv
source .venv/bin/activate

pip install --upgrade pip
pip install pyspark==4.2.0

python3 -c "
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local[*]').appName('verify-pyspark').getOrCreate()
print('PySpark is running. Spark version:', spark.version)
spark.stop()
"

# 1. IMPORTS - bring in only what you need
from pyspark.sql import SparkSession
from pyspark.sql import functions as F   # common alias for built-in column functions

# 2. ENTRY POINT - one SparkSession per script, built once
spark = (SparkSession.builder
    .master("local[*]")          # run locally, use all available cores
    .appName("my-script-name")   # shows up in the Spark UI / logs
    .getOrCreate())
sc = spark.sparkContext          # only needed if you use the older RDD API

# 3. HOUSEKEEPING - quiet the noisy default logger (optional, but recommended)
sc.setLogLevel("ERROR")

# 4. LOAD DATA - read a file, or build a small dataset from Python values
df = spark.read.csv("data/some_file.csv", header=True, inferSchema=True)

# 5. TRANSFORM - describe the pipeline; nothing runs yet (lazy evaluation)
result = (df
    .filter(F.col("status") == "delivered")      # keep only relevant rows
    .groupBy("category")                         # MAP/SHUFFLE step
    .agg(F.count("*").alias("total")))            # REDUCE step

# 6. ACTION - this is what actually triggers computation across the cluster
result.show()          # or .collect(), .write.csv(...), .count(), etc.

# 7. CLEAN UP - release the cluster resources when you're done
spark.stop()

from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[*]").appName("word-count").getOrCreate()
sc = spark.sparkContext
sc.setLogLevel("ERROR")   # silence the WARN noise from Step 1

text = """Welcome to the Big World of Big Big Data Welcome World bye
World Hello MapReduce GoodBye MapReduce
This Book on Big Data is fun"""

lines = sc.parallelize(text.strip().split("\n"))   # one RDD partition per line

word_counts = (lines
    .flatMap(lambda line: line.split(" "))   # MAP: line -> many words (flatMap, not map!)
    .map(lambda word: (word, 1))             # MAP: word -> (word, 1)
    .reduceByKey(lambda a, b: a + b))        # SHUFFLE + REDUCE: sum the 1s per word

for word, count in sorted(word_counts.collect()):
    print(f"{word}: {count}")

spark.stop()

python3 word_count.py

import csv
import io
from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[*]").appName("orders-per-month").getOrCreate()
sc = spark.sparkContext
sc.setLogLevel("ERROR")

DATA_PATH = "data/olist_dataset/olist_orders_dataset.csv"

raw = sc.textFile(DATA_PATH)
header = raw.first()
rows = raw.filter(lambda line: line != header)   # drop the CSV header row

def parse_line(line):
    return next(csv.reader(io.StringIO(line)))   # handles quoted commas correctly

parsed = rows.map(parse_line)

# order_purchase_timestamp is column index 3, e.g. "2017-10-02 10:56:33"
orders_per_month = (parsed
    .map(lambda cols: (cols[3][:7], 1))   # MAP: (k, v) -> ("2017-10", 1)
    .reduceByKey(lambda a, b: a + b)      # SHUFFLE + REDUCE: sum per month
    .sortByKey())                         # chronological order

for month, count in orders_per_month.collect():
    print(f"{month}: {count} orders")

spark.stop()

python3 orders_per_month.py

from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[*]").appName("combiner-demo").getOrCreate()
sc = spark.sparkContext
sc.setLogLevel("ERROR")

text = """Welcome to the Big World of Big Big Data Welcome World bye
World Hello MapReduce GoodBye MapReduce
This Book on Big Data is fun"""
pairs = (sc.parallelize(text.strip().split("\n"))
    .flatMap(lambda line: line.split(" "))
    .map(lambda w: (w, 1)))

# groupByKey: ships every single (word, 1) pair across the network first,
# THEN sums each key's full list locally. No local pre-aggregation.
grouped = pairs.groupByKey().mapValues(sum)

# reduceByKey: combines same-key pairs on each partition FIRST (the combiner),
# and only sends already-summed partial totals across the network.
combined = pairs.reduceByKey(lambda a, b: a + b)

print("groupByKey :", sorted(grouped.collect()))
print("reduceByKey:", sorted(combined.collect()))
print("Identical results:", sorted(grouped.collect()) == sorted(combined.collect()))

spark.stop()

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder.master("local[*]").appName("revenue-by-category").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

DATA = "data/olist_dataset"

order_items = spark.read.csv(f"{DATA}/olist_order_items_dataset.csv", header=True, inferSchema=True)
products = spark.read.csv(f"{DATA}/olist_products_dataset.csv", header=True, inferSchema=True)
categories = spark.read.csv(f"{DATA}/product_category_name_translation.csv", header=True, inferSchema=True)

revenue_by_category = (
    order_items
    .join(products, on="product_id", how="inner")
    .join(categories, on="product_category_name", how="left")
    .groupBy("product_category_name_english")
    .agg(
        F.round(F.sum("price"), 2).alias("total_revenue"),
        F.count("order_id").alias("items_sold")
    )
    .orderBy(F.desc("total_revenue"))
)

revenue_by_category.show(10, truncate=False)

spark.stop()

python3 revenue_by_category.py

from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression
from pyspark.ml import Pipeline
from pyspark.ml.evaluation import RegressionEvaluator

spark = SparkSession.builder.master("local[*]").appName("predict-shipping-cost").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

DATA = "data/olist_dataset"

order_items = spark.read.csv(f"{DATA}/olist_order_items_dataset.csv", header=True, inferSchema=True)
products = spark.read.csv(f"{DATA}/olist_products_dataset.csv", header=True, inferSchema=True)

df = (order_items.join(products, on="product_id", how="inner")
      .select("price", "freight_value", "product_weight_g",
              "product_length_cm", "product_height_cm", "product_width_cm")
      .dropna())

# 80/20 train/test split. seed=42 makes the split reproducible.
train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)

feature_cols = ["price", "product_weight_g", "product_length_cm",
                "product_height_cm", "product_width_cm"]

assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
lr = LinearRegression(featuresCol="features", labelCol="freight_value", maxIter=10, regParam=0.3)
pipeline = Pipeline(stages=[assembler, lr])

# fit() is an Estimator call -> produces a PipelineModel (a Transformer)
model = pipeline.fit(train_df)

# transform() applies the fitted pipeline to unseen test data
predictions = model.transform(test_df)
predictions.select("freight_value", "prediction").show(5)

evaluator = RegressionEvaluator(labelCol="freight_value", predictionCol="prediction", metricName="rmse")
rmse = evaluator.evaluate(predictions)
print(f"RMSE: {rmse:.2f}")

lr_model = model.stages[1]   # the fitted LinearRegressionModel inside the pipeline
print("Coefficients:", lr_model.coefficients)
print("Intercept:", lr_model.intercept)

spark.stop()

python3 predict_shipping_cost.py