Design a reproducible data pipeline for modeling

You receive raw clickstream events and a user table. Build a reproducible daily pipeline that outputs user-day features for modeling. It must be idempotent, handle late-arriving/out-of-order events, include quality checks and unit tests, and support backfills.

Schema and small samples:

users
user_id | signup_date | country
u1 | 2025-05-20 | US
u2 | 2025-06-02 | CA
u3 | 2025-06-15 | US

events
event_id | user_id | ts | event_type | amount_usd | source
e1 | u1 | 2025-06-01T12:00:00Z | page_view | null | ads
e2 | u1 | 2025-06-01T12:10:00Z | add_to_cart | null | ads
e3 | u1 | 2025-06-01T12:45:00Z | purchase | 30.0 | ads
e4 | u2 | 2025-06-02T00:05:00Z | page_view | null | seo
e5 | u2 | 2025-06-02T00:40:00Z | purchase | 20.0 | seo

Tasks:

1. In SQL or Python, define transformations to produce user-day features: sessions (30-min inactivity gap), total_page_views, add_to_cart_count, purchases_count, revenue_usd, days_since_signup, country, and top_source for the day; dedupe by event_id; ensure UTC→date bucketing is correct.
2. Describe how you guarantee idempotency and correctness with late/out-of-order data (e.g., watermarking, upserts/merge, partition overwrite vs. append-only with versioning).
3. Specify a backfill plan for 2025-06-01 to 2025-08-31, including how you would re-run only affected partitions safely.
4. Propose concrete data-quality checks (row-count reconciliations, not-null/valid-set for event_type, nonnegative revenue) and two unit tests that would have caught common bugs.
5. Outline orchestration (DAG tasks and dependencies), storage formats/partitioning, and how you would expose the output for both training and online inference.

Build daily user features with sessions from clickstream events

You are given two tables, users and events, with raw clickstream data. For each user and calendar day on which they generated at least one event, produce a single row of user-day features.

Definitions and requirements:

• A "day" is based on the UTC timestamp ts, using the calendar date of ts.
• A "session" is a contiguous sequence of events for a user on a given day where the gap between consecutive events is at most 30 minutes. A new session starts if the previous event is more than 30 minutes earlier or does not exist (first event of the day).
• If duplicate rows exist for the same event_id, they should be de-duplicated so that each event_id contributes at most once.
• For each user_id and day, output the following columns:
  • user_id
  • event_date (the UTC calendar date of ts)
  • country (from the users table)
  • sessions (number of sessions that day)
  • total_page_views (number of events with event_type = 'page_view')
  • add_to_cart_count (number of events with event_type = 'add_to_cart')
  • purchases_count (number of events with event_type = 'purchase')
  • revenue_usd (sum of amount_usd for purchase events on that day; treat null as 0)
  • days_since_signup (event_date minus signup_date, in whole days)
  • top_source (the source value with the highest number of events for that user and day; if there is a tie, pick the lexicographically smallest source)

Write a single SQL query that produces this user-day feature table from the schemas and sample data below.

Table users

column	type	constraints
user_id	VARCHAR(10)	PRIMARY KEY
signup_date	DATE	NOT NULL
country	VARCHAR(2)	NOT NULL

Sample data:

country	signup_date	user_id
US	2025-05-20	u1
CA	2025-06-02	u2
US	2025-06-15	u3

Table events

column	type	constraints
event_id	VARCHAR(10)	PRIMARY KEY
user_id	VARCHAR(10)	NOT NULL
ts	TIMESTAMP	NOT NULL
event_type	VARCHAR(50)	NOT NULL
amount_usd	DECIMAL(10,2)
source	VARCHAR(20)	NOT NULL

Sample data:

amount_usd	event_id	event_type	source	ts	user_id
NULL	e1	page_view	ads	2025-06-01T12:00:00	u1
NULL	e2	add_to_cart	ads	2025-06-01T12:10:00	u1
30	e3	purchase	ads	2025-06-01T12:45:00	u1
NULL	e4	page_view	seo	2025-06-02T00:05:00	u2
20	e5	purchase	seo	2025-06-02T00:40:00	u2

Expected output

One row per (user_id, event_date) with daily session counts, per-event-type counts, revenue, days since signup, country, and top_source for that day, for users that have at least one event on that date.

add_to_cart_count	country	days_since_signup	event_date	purchases_count	revenue_usd	sessions	top_source	total_page_views	user_id
1	US	12	2025-06-01	1	30	2	ads	1	u1
0	CA	0	2025-06-02	1	20	2	seo	1	u2

Hints

1. Start by de-duplicating events on event_id using ROW_NUMBER() over PARTITION BY event_id.
2. Use window functions (LAG) to detect 30-minute gaps for session boundaries, and conditional aggregation (FILTER or CASE) to compute per-event-type counts.