Explain a complex Python PySpark code you have created in either your current or past projects. First describe the business problem, then explain the code you wrote to solve it.

In my previous role at a digital marketing firm, we needed to generate real‑time engagement metrics from millions of click‑stream events. The business problem was to calculate per‑user session durations, page‑view counts, and conversion rates within a 5‑minute window, and to surface anomalies for the analytics team. I designed a PySpark job that ingested raw JSON logs from Kafka, performed schema validation, and used Spark’s Structured Streaming API. The core logic involved windowed aggregations: a tumbling window of 5 minutes, a sliding window of 1 minute, and a session window with a 30‑second inactivity gap. I leveraged Spark SQL functions such as window, count, avg, and collect_list to compute metrics, and used when/otherwise to flag sessions exceeding a threshold. The job wrote results to a Delta Lake table for downstream BI tools. Performance tuning included partitioning by hour, caching intermediate DataFrames, and adjusting the spark.sql.shuffle.partitions setting. The solution processed 10 million events per hour with a latency of under 30 seconds, meeting the SLA. This example demonstrates my ability to translate a business requirement into a scalable PySpark pipeline, optimize for performance, and ensure data quality.