Introduction
The shop can now distinguish a viewed session/product pair from repeated events. Its next question is which products have a large share of viewed sessions without a valid cart, and how much evidence supports each comparison.
You will produce a complete table with the viewed-session denominator, completed count, incomplete count and incomplete proportion. Keep products with zero valid carts and every tied result. The result is an observation clue for further investigation, not proof that a product or page caused a visit to stop.
This independent lab supplies the completed journey module from the preceding guided lab and fresh raw files. You should know that lab's earliest-view, session/product and time rules, plus Spark grouping and Parquet. Use Terminal. The batch is finite and closed; identifiers and UTC second timestamps are valid, and repeated event IDs are exact copies. The author estimates 35–45 minutes. There is no streaming, unique-person estimate or causal inference.
Build the Complete Product Comparison
In this step, you will aggregate one row per viewed journey into one row per product. Start in Terminal. Among the supplied programs, edit only analysis.py. Preserve the raw files and supplied run_analysis.py, read_result.py and support.py unchanged; derived results belong under output/:
cd /home/labex/project
cat data/batch.csv
cat analysis.py
This module contains read_events, build_views and build_journeys, independently prepared from the preceding lab. A module holds reusable functions. The driver supplies the current SparkSession and input path, then passes the returned journey DataFrame to your new build_report(journeys) function. Nothing from a previous VM or terminal session is required.
Recall the contract: identical full event copies are removed; each viewed (session_id, product_id) uses its earliest view; a cart is valid only for the same two keys and at or after that view. Several valid carts still produce one completed journey. The driver uses local[2], UTC formatting, and the same Python for driver and workers. It writes outputs after your functions return.
For a product with three viewed pairs and two completed pairs, hand-calculate three viewed sessions, two cart sessions, one no-cart session and a no-cart ratio of 1/3. A ratio without that denominator could make a single visit look as convincing as a large sample.
count("*") now counts journey rows, not raw events. Casting a Boolean completed to long maps true to 1 and false to 0. Spark's sum adds these values. Subtracting the completed count gives the incomplete count; dividing by the positive viewed-session count gives a ratio between zero and one. Every group here has at least one view. A product with only carts has no group or denominator; we do not invent a zero-denominator ratio.
Write Only the Product Aggregation
Keep the supplied read_events, build_views and build_journeys: they are the completed work from the preceding lab. Do not retype them. F is their existing alias for Spark functions. Open the same module with nano:
nano analysis.py
Use the arrow keys to reach the end of build_journeys, then leave a blank line. As practiced, Ctrl+W searches for a function name; each new declaration starts at the left margin and its body uses four spaces. You will write the complete build_report, not fill a prepared answer template.
First group the journey DataFrame by product. Derive two counts in that same aggregation: all rows for the viewed denominator, then the sum of Boolean completion cast to an integer. Keep the field names used by the driver:
def build_report(journeys):
counts = journeys.groupBy("product_id").agg(
F.count("*").alias("view_sessions"),
F.sum(F.col("completed").cast("long")).alias("cart_sessions"),
)
Continue within your function. withColumn derives a named column from the grouped counts. Subtract completed from viewed before dividing by that same positive viewed denominator. Do not filter on completed first: doing so would erase products with no valid carts. Do not select only one maximum: tied products all belong in the report.
return counts.withColumn(
"no_cart_sessions", F.col("view_sessions") - F.col("cart_sessions")
).withColumn(
"no_cart_ratio", F.col("no_cart_sessions") / F.col("view_sessions")
)
Save with Ctrl+O, confirm analysis.py with Enter, then exit with Ctrl+X. Reopen to inspect your saved function and verify that the earlier functions remain:
nano analysis.py
Use Ctrl+W to find build_report. For a mistake, use arrows and Backspace/Delete to replace only the affected line; do not add a second definition. Keep the continuation indentation and closing parentheses. Save changes, exit and read the file:
cat analysis.py
If the next run reports SyntaxError or IndentationError, locate the named line in nano, check the surrounding function, parentheses and four-space indentation, save and rerun. A successfully computed but wrong count or ratio calls for checking the grouping, cast and denominator instead. Do not reset the VM or rewrite the working G1 functions to repair your report.
Run the report stage with explicit input and output paths:
bigdata-python run_analysis.py data/batch.csv output/main --stage report
Expect this complete logical table; displayed floating values may have more digits:
product_id view_sessions cart_sessions no_cart_sessions no_cart_ratio
card 1 0 1 1.0
pin 1 0 1 1.0
tea 3 1 2 0.6666666666666666
book 2 1 1 0.5
mug 3 2 1 0.3333333333333333
The driver reads the actual saved Parquet before presenting it. Its table sorts by descending no_cart_ratio, then product identifier only to make ties easy to read. Both card and pin remain; a sort tie is not permission to discard either. The poster cart-only product is excluded because it had no viewed pair.
The results are typed datasets at output/main/journeys and output/main/report. The latter has a string product identifier, whole-number counts and a numeric ratio. Parquet part filenames and their number are implementation details, not the report's identity. Equivalent Spark DataFrame/SQL expressions are accepted; retain a pipeline from the supplied input and keep transformations inside Spark. Do not collect raw journeys to Python, switch to RDD/pandas, or checkpoint and recreate them.
Consume the Files and Rebuild the Snapshot
In this step, you will consume the small CSV and replace the derived snapshot without changing its source. Continue in Terminal at /home/labex/project.
The prepared driver saves output/main/report.csv as a single small UTF-8 CSV for a colleague. It collects only the final, already aggregated product table for serialization, after build_report returns. This final presentation does not replace the Spark transformations. View the real file:
cat output/main/report.csv
The header has the same five fields as the Parquet report. Check all five products, the zero cart counts for card and pin, and both tied ratios of 1.0. A CSV can be read as rows and fields; counting lines or trusting its filename would not prove these relationships.
Use the provided reader to consume both real Parquet datasets and the CSV. read_result.py reads the supplied output path, formats timestamps inside Spark in UTC, and compares the CSV's content with the stored report; it does not create or fill missing results:
bigdata-python read_result.py output/main
Expect ten journey rows and five product rows, plus CSV matches stored report. If a file is missing, rerun your analysis with the correct input and destination; the reader intentionally fails rather than inventing an answer.
sha256sum prints a content fingerprint for each raw file. The *.csv pattern selects the CSV files in data/. Compare these lines before and after rebuilding:
sha256sum data/*.csv
bigdata-python run_analysis.py data/batch.csv output/main --stage report
bigdata-python read_result.py output/main
sha256sum data/*.csv
The same input gives the same logical report without duplicate products or journeys. The raw fingerprints must be unchanged. New Parquet filenames or bytes can vary between builds; logical content is the repeatability contract. The writer replaces only this known derived destination, not the raw files or another result directory.
Now use the saved module with the normal sample and a separate destination:
bigdata-python run_analysis.py data/normal.csv output/normal --stage report
bigdata-python read_result.py output/normal
Expect two product rows, each with one viewed session, one cart session, zero no-cart sessions and ratio 0. The main result remains intact. The same parameterized functions support both batches.
card and pin deserve attention in this tiny batch, but each has only one observed session. A high ratio does not establish why a visitor stopped, identify unique people, or guarantee the pattern will persist. Counts and complete ties make the observation reviewable; further evidence would be needed for a causal claim.
Grading reads your actual live results without rebuilding them. It also loads the unchanged module and calls its functions on a disposable batch with different identifiers and row order, observes the real Spark input lineage and consumes disposable outputs. It does not score an exact command spelling, self-written plan or application-name marker, and it never repairs your live result. Keep the functions free of writes; the driver owns serialization.
Summary
You built and consumed complete Parquet and CSV comparisons, retained zero-cart products and ties, and rebuilt a derived snapshot while preserving raw inputs. The denominator and count columns explain each ratio. These closed-session patterns support further observation, not a claim of causation.
