Build the pipeline

# clean_job.sh: bronze -> cleaned text with Hadoop Streaming (3 reducers)
hdfs dfs -rm -r -f -skipTrash /capstone/stage/clean
mapred streaming \
  -D mapreduce.job.name=capstone-clean \
  -D mapreduce.job.reduces=3 \
  -files clean_mapper.py,clean_reducer.py,capstone_raw/menu.csv \
  -input '/capstone/bronze/orders/college=*' \
  -output /capstone/stage/clean \
  -mapper clean_mapper.py -reducer clean_reducer.py

# time bash clean_job.sh      (clean_mapper.py and clean_reducer.py: see Lesson 14.1)

# silver_gold.py: cleaned text -> silver Parquet (partitioned by day) -> gold KPI tables
import sys
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.appName("capstone-silver-gold")
         .config("spark.sql.sources.partitionOverwriteMode", "dynamic")   # re-runs replace only their days
         .config("spark.sql.shuffle.partitions", "6").getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
B = "hdfs://namenode/capstone"
cols = ["status", "order_id", "dt", "ts", "college", "student_id", "item", "category", "qty", "amount",
        "pay_mode", "late", "unknown_item", "undercharged"]

schema = ", ".join(c + " STRING" for c in cols)          # read everything as text first, then cast
raw = spark.read.schema(schema).option("sep", "\t").csv(f"{B}/stage/clean")   # BAD rows fill 3 columns
orders = (raw.where("status = 'OK'").drop("status")
          .withColumn("ts", F.to_timestamp("ts")).withColumn("dt", F.to_date("dt"))
          .withColumn("qty", F.col("qty").cast("int")).withColumn("amount", F.col("amount").cast("int"))
          .withColumn("late", F.col("late") == "1").withColumn("unknown_item", F.col("unknown_item") == "1")
          .withColumn("undercharged", F.col("undercharged").cast("int")))
days = sys.argv[1:]                                    # optional: only these days (a backfill)
if days:
    orders = orders.where(F.col("dt").isin(days))
orders.write.mode("overwrite").partitionBy("dt").parquet(f"{B}/silver/orders")
raw.where("status = 'BAD'").select("order_id", "dt").toDF("file_day", "raw_line") \
   .write.mode("overwrite").json(f"{B}/quarantine/orders")

pay = spark.read.json(f"{B}/bronze/payments").withColumn("dt", F.to_date(F.substring("ts", 1, 10)))
pay.write.mode("overwrite").partitionBy("dt").parquet(f"{B}/silver/payments")

o = spark.read.parquet(f"{B}/silver/orders")
p = spark.read.parquet(f"{B}/silver/payments")
ok = p.where("status = 'SUCCESS'").select("order_id", F.col("amount").alias("paid"))
upi = o.where("pay_mode = 'UPI'").join(ok, "order_id", "left")          # every UPI order, paid or not
gold = {
  "daily_college": o.groupBy("dt", "college").agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue")),
  "item_mix": o.groupBy("college", "item", "category").agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue")),
  "hourly": o.groupBy(F.hour("ts").alias("hour")).agg(F.count("*").alias("orders")),
  "upi_recon": (o.where("pay_mode = 'UPI'").groupBy("college").agg(F.sum("amount").alias("pos_upi"))
                .join(p.join(o.select("order_id", "college"), "order_id").groupBy("college").agg(
                      F.sum(F.when(F.col("status") == "SUCCESS", F.col("amount")).otherwise(0)).alias("gateway_success"),
                      F.sum(F.when(F.col("status") == "REFUNDED", F.col("amount")).otherwise(0)).alias("refunds"),
                      F.sum(F.when(F.col("status") == "FAILED", 1).otherwise(0)).alias("failed_attempts")), "college")
                .withColumn("gap", F.col("pos_upi") - F.col("gateway_success"))),
  "upi_gaps": (upi.where("paid IS NULL").groupBy("dt", "college", F.hour("ts").alias("hour"))
               .agg(F.count("*").alias("orders"), F.sum("amount").alias("unconfirmed"))),
}
for name, df in gold.items():
    df.coalesce(1).write.mode("overwrite").parquet(f"{B}/gold/{name}")

print("== silver.orders ==")
o.agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue"), F.countDistinct("dt").alias("days"),
      F.sum(F.col("late").cast("int")).alias("late"), F.sum(F.col("unknown_item").cast("int")).alias("unknown"),
      F.sum("undercharged").alias("undercharged")).show()
print("== gold.daily_college: month by canteen ==")
spark.read.parquet(f"{B}/gold/daily_college").groupBy("college").agg(F.sum("orders").alias("orders"),
      F.sum("revenue").alias("revenue")).orderBy(F.desc("revenue")).show()
print("== gold.upi_recon ==")
spark.read.parquet(f"{B}/gold/upi_recon").orderBy("college").show()
print("== gold.upi_gaps: UPI orders with no SUCCESS from the gateway ==")
spark.read.parquet(f"{B}/gold/upi_gaps").orderBy("dt", "college", "hour").show()
print("== gold.hourly: top 3 hours ==")
spark.read.parquet(f"{B}/gold/hourly").orderBy(F.desc("orders")).show(3)
print("== gold.item_mix: maggi and Panvel chai ==")
spark.read.parquet(f"{B}/gold/item_mix").where("item = 'maggi' or (college = 'PNL' and item = 'chai')").show()

# spark_job.sh: run silver_gold.py on YARN (pass days to backfill only those days)
export SPARK_HOME=/tmp/opt/spark-4.0.4-bin-hadoop3 HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
$SPARK_HOME/bin/spark-submit --master yarn --deploy-mode client \
  --num-executors 3 --executor-memory 768m --executor-cores 1 --driver-memory 768m \
  silver_gold.py "$@"

# check.py: acceptance tests. The pipeline's answers must equal truth.json from the generator.
import json
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("capstone-check").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
B = "hdfs://namenode/capstone"
t = json.load(open("capstone_raw/truth.json"))
o = spark.read.parquet(f"{B}/silver/orders")
s = o.agg(F.count("*"), F.sum("amount"), F.countDistinct("order_id"), F.sum(F.col("late").cast("int")),
          F.sum(F.col("unknown_item").cast("int")), F.sum("undercharged")).first()
recon = spark.read.parquet(f"{B}/gold/upi_recon").agg(F.sum("gateway_success"), F.sum("refunds"), F.sum("gap")).first()
daily = spark.read.parquet(f"{B}/gold/daily_college").agg(F.sum("revenue")).first()[0]
tests = [
    ("orders = truth", s[0], t["clean_orders"]),
    ("revenue = truth", s[1], t["revenue"]),
    ("order_id is unique", s[2], s[0]),
    ("late orders kept", s[3], t["late"]),
    ("unknown items flagged", s[4], t["unknown_item"]),
    ("undercharged rupees", s[5], t["undercharged"]),
    ("quarantined lines", spark.read.json(f"{B}/quarantine/orders").count(), t["malformed"]),
    ("gold revenue = silver", daily, s[1]),
    ("UPI net = truth", recon[0] + recon[1], t["upi_net"]),
    ("UPI gap = unconfirmed", recon[2], t["upi_unconfirmed"]),
]
for name, got, want in tests:
    print(f"{'PASS' if got == want else 'FAIL'}  {name:<24} got {got:>10,}  expected {want:>10,}")
print(f"{sum(g == w for _, g, w in tests)}/{len(tests)} checks passed")

export SPARK_HOME=/tmp/opt/spark-4.0.4-bin-hadoop3 HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
$SPARK_HOME/bin/spark-submit --master yarn --num-executors 2 --executor-memory 768m --driver-memory 768m check.py

# idem.sh: backfill one day twice, and show what blind "append" would have done
export SPARK_HOME=/tmp/opt/spark-4.0.4-bin-hadoop3 HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
S="$SPARK_HOME/bin/spark-submit --master yarn --num-executors 3 --executor-memory 768m --driver-memory 768m"
q() { hdfs dfs -ls -R /capstone/silver/orders/dt=2026-09-14 | grep -c parquet; }
echo "files in dt=2026-09-14 before: $(q)"
$S silver_gold.py 2026-09-14 2>/dev/null | sed -n '/silver.orders/,/^$/p'
$S silver_gold.py 2026-09-14 2>/dev/null | sed -n '/silver.orders/,/^$/p'
echo "files in dt=2026-09-14 after two backfills: $(q)"
$S append_demo.py 2>/dev/null

# append_demo.py: the same day written twice with mode("append") vs mode("overwrite")
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("capstone-append-demo").getOrCreate()
day = spark.read.parquet("hdfs://namenode/capstone/silver/orders").where("dt = '2026-09-14'")
for mode in ["append", "overwrite"]:
    path = f"hdfs://namenode/capstone/tmp/{mode}_demo"
    for run in (1, 2):
        day.write.mode(mode).parquet(path)
        r = spark.read.parquet(path).agg(F.count("*"), F.sum("amount")).first()
        print(f"{mode:<9} run {run}: orders {r[0]:,}  revenue {r[1]:,}")