Background
On a grey Saturday morning in Paris, Chloé Laurent, the Lead Data Engineer at SantéFlux, sat in her office
overlooking the Seine. She had just received a sharp email from the Chief Privacy Officer (CPO). The subject
line read: "Urgent: Analytics Access Blocked."
SantéFlux is a top French health-tech company. Their main tool, "Stress Spikes," finds times when a user’s
heart rate stays above 100 BPM for over 10 minutes while they are resting. This feature is vital for new
medical partnerships across Europe. However, Chloé’s team was now locked out of the data. The company's
future was at risk.
The SantéFlux platform gets millions of 'bio-pings' every day from smart watches. Every ping includes a
metadata tag for the user's City (such as Paris, Lyon, or Marseille) based on their registered home region.
While the data is great for health study, it is a legal risk. To keep the project alive, Chloé must transform
these pings into a 'City Stress Matrix', a report that groups heart rate trends by location and subscription
tier. However, the raw data is far too personal for the legal team, and current reports are taking too long to
run. Even with the dataset capped at 85 MB for testing, it holds millions of records that must be handled
with care.
The Privacy Crisis
The CPO, Jean-Luc Moreau, found full names and raw IDs in shared folders. "Until this data is masked,
hashed with secret keys, and grouped at a safe level, the Stress Spikes project is dead," Moreau wrote.
Chloé had a hard choice. She had to give a Health Trends Report showing average heart rates for cities
like Paris, Lyon, and Marseille while following Moreau’s strict GDPR rules. Her team planned a "Privacy
Shield" pipeline, but it led to new choices:
• PII Masking: Should the team hide names only at the very end to keep things simple for now? Or
should they scrub the raw data right away to stop any risk, even if it makes fixing bugs harder later?
• Salted Hashing: Moreau wanted "Salting" adding a secret string to IDs before hashing them. Chloé
knew this would stop hackers, but it would also make joining data with other teams very hard if they
did not share the same secret string.
• The Frozen Sensor: Checks showed that some sensors were "freezing," sending the same heart rate
for hours (see Exhibit 2). If Chloé’s team removed these rows too fast, they might lose real data
from stable users. If they kept them, the city averages would be wrong.
Workspace
Actionables
Importing all the functions and datatypes of pyspark library
StructType and StructField defines a hardcoded schema. Instead of letting Spark "guess" the data types (which can be slow and error-prone), we are explicitly telling it that UserID is an integer and SubscriptionType is a string. And then we sort the resulting DataFrame by UserID.
StructType Schema defines six specific columns required for the health analysis. Also it uses TimestampType for the time of the ping and DoubleType for the HeartRate to ensure precise calculations for the "Stress Spikes" logic. The schema also includes ville (the French metadata tag for City) and Status, which Chloé will use to identify the "FROZEN" sensor data.
‘shared_salt’ defines the secret string used to protect user IDs from being easily reverse-engineered. ‘regexp_replace’ masks the FullName column by keeping the first name and the first initial of the last name, replacing the rest with asterisks (e.g., "Pierre D****"). sha2 and concat creates a SecureID by combining the UserID with the shared_salt and hashing it using the SHA-256 algorithm.
window("Timestamp", "10 minutes") function creates a new column that assigns each "bio-ping" to a specific 10-minute time block based on its timestamp. This is the exact duration required by the business logic to identify spikes. “Time_Window" column contains start and end times for each 10-minute interval, allowing for time-series aggregation later. Then we sort by the salted hash ID and time making it easier to visually verify a single user's heart rate progression across consecutive windows.
groupBy function organizes the data by each individual user and their specific 10-minute time block to analyze heart rate stability over time. .agg calculates both the min and max heart rate recorded for that user during that specific window.
.filter(col("min_rate") == col("max_rate") is the core logic for finding bad data; if the minimum and maximum values are identical, the sensor was likely "frozen" and not recording real biological changes.
join(..., how="leftanti") is a specialized join type used for filtering. A Left Anti Join keeps only the rows from the left table (df_window) that do not have a matching record in the right table (frozen_hb_df). on=["SecureID","Time_Window"] ensures the filtering is precise. It looks for the exact combination of a specific user and the specific 10-minute window identified as "frozen" in the previous step.
shared_salt uses the exact same secret string defined in the previous step. For a join to work, the "salt" must be identical across both tables so the resulting hashes match. sha2(concat(...)) combines the raw UserID with the salt and hashes it using the SHA-256 algorithm. This satisfies the CPO's requirement to mask and hash data before it is shared.
.drop("UserID") removes the original, unencrypted ID from the DataFrame to ensure the team is only working with "Secure IDs".
groupBy("SecureID") groups the data by the salted hash ID. This is a safeguard in case a user has multiple records (for example, if they upgraded from free to premium during the testing period).
By using the max function in .agg(max("SubscriptionType")...), Chloé picks a single subscription type for each user. In most systems, "Premium" or "VIP" would alphabetically or logically override "Free," ensuring the most current or highest-tier status is captured.
.alias("Subscription_Type") renames the resulting column for better readability in the final "City Stress Matrix" report.
join(..., "leftanti") executes a Left Anti Join, which returns only the rows from the clean vitals data (vitals_clean_df) where the SecureID does not exist in the subscription table (df_subs_masked). "SecureID" is the join key; since both tables were hashed using the same "shared salt," Chloé can accurately match users across datasets without seeing their real names.
join(..., "leftanti") performs a Left Anti Join, which isolates records in the vitals_clean_df that do not have any matching SecureID in the df_subs_final table. By joining against the finalized subscription table (which we previously deduplicated and aggregated), Chloé ensures she is checking against the most accurate list of known users.
join(..., "inner") is an Inner Join, which discards any "Ghost Users" (vitals without subscriptions) and any subscribers who haven't sent any heart rate pings. "SecureID" is the common link between the two tables, which was previously hashed and salted to comply with the CPO's privacy rules.
broadcast(df_subs_final) is the key optimization. Since the subscription table is only about 125 KB, Spark sends a copy of it to every executor. This prevents a "Shuffle," which Chloé suspected was causing the system to crash or run indefinitely. "inner" join type ensures that only users with both heart rate data and a verified subscription tier are included in the board's report, effectively filtering out "Ghost Users".
groupBy("City") organizes the data by geographic location (e.g., Paris, Lyon, Marseille) to meet the requirement for a location-based health report.
.pivot("Subscription_Type") transformation turns the distinct values in the subscription column (like "Abonnement Premium" and "Free") into individual columns. This allows the Board to compare the two groups side-by-side.
.agg(...) calculates the specific metrics needed for the "Stress Spikes" analysis, the average heart rate to see general trends and the maximum heart rate to identify peak stress moments.
