Data Transformation and Loading Questions

create or replace table stg_delta as
select s.pk,
       s.*,
       hash_md5(concat_ws('|', s.c1, s.c2, s.c3)) as new_hash
from (
  select distinct on (pk) * from stg_raw  -- or use ROW_NUMBER() to dedupe
) s
left join target t on s.pk = t.pk
where t.pk is null OR hash_md5(concat_ws('|', t.c1, t.c2, t.c3)) <> hash_md5(concat_ws('|', s.c1, s.c2, s.c3));

merge into target tgt
using (
  select * from stg_delta where mod(abs(hash_md5(pk)::bigint), 100) between 0 and 9
) src
on src.pk = tgt.pk
when matched then update set c1 = src.c1, c2 = src.c2, ...
when not matched then insert (pk, c1, c2, ...) values (src.pk, src.c1, src.c2, ...);

create or replace table target_new as
select coalesce(src.pk, tgt.pk) as pk, coalesce(src.c1, tgt.c1) as c1, ...
from target tgt
full outer join (
  select * from stg_delta where date_col between '2025-11-01' and '2025-11-07'
) src on src.pk = tgt.pk
where -- keep only rows for unaffected partitions + merged results
;

alter table target rename to target_old;
alter table target_new rename to target;  -- fast metadata swap
drop table if exists target_old;

WITH ordered AS (
  SELECT user_id, event_ts,
         LAG(event_ts) OVER (PARTITION BY user_id ORDER BY event_ts) AS prev_ts
  FROM events
  WHERE event_ts >= TIMESTAMP_SUB(CURRENT_TIMESTAMP, INTERVAL 7 DAY) -- scope
),
flags AS (
  SELECT *,
    CASE WHEN prev_ts IS NULL OR TIMESTAMP_DIFF(event_ts, prev_ts, MINUTE) > 30 THEN 1 ELSE 0 END AS new_session
  FROM ordered
),
sess AS (
  SELECT user_id, event_ts,
    SUM(new_session) OVER (PARTITION BY user_id ORDER BY event_ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS session_seq
  FROM flags
)
SELECT user_id,
       CONCAT(user_id, '_', session_seq) AS session_id,
       MIN(event_ts) AS start_ts,
       MAX(event_ts) AS end_ts,
       TIMESTAMP_DIFF(MAX(event_ts), MIN(event_ts), SECOND) AS duration_s
FROM sess
GROUP BY user_id, session_seq;

KeyedStream<Event, userId> keyed = events.assignTimestampsAndWatermarks(wmStrategy);
keyed.process(new KeyedProcessFunction() {
  MapState<String, SessionState> sessions;
  onEvent(e):
    find candidate session(s) within [e.ts - 30m, e.ts + 30m]
    if merges required:
      delete old sessions -> emit correction DELETEs
      create merged session -> emit INSERT or UPDATE with before/after
      upsert sessions store
    else extend single session -> emit UPDATE correction
  onTimer(watermark passes session_end + allowed_lateness):
    mark session.is_final = true
    optionally emit final session (no more corrections expected)
});

import pandas as pd

def etl_csv_to_parquet(csv_path: str, parquet_path: str):
    """
    Reads CSV with columns [user_id, signup_ts, revenue],
    cleans and transforms, deduplicates by user_id keeping the most recent signup_ts,
    and writes result to parquet.
    """
    # Read CSV (for moderate files)
    df = pd.read_csv(csv_path, dtype={"user_id": str, "signup_ts": str, "revenue": str})

    # Parse signup_ts as UTC, convert to America/New_York timezone
    df['signup_ts'] = pd.to_datetime(df['signup_ts'], utc=True, errors='coerce')
    df['signup_ts'] = df['signup_ts'].dt.tz_convert('America/New_York')

    # Clean revenue: remove $ and commas, convert to float. Coerce errors to NaN.
    df['revenue'] = (
        df['revenue']
        .str.replace(r'[\$,]', '', regex=True)
        .astype(float, errors='ignore')
    )
    df['revenue'] = pd.to_numeric(df['revenue'], errors='coerce')

    # Drop rows missing user_id or signup_ts (or handle as business rule)
    df = df.dropna(subset=['user_id', 'signup_ts'])

    # Deduplicate: keep row with max signup_ts per user_id
    df = df.sort_values('signup_ts').drop_duplicates(subset='user_id', keep='last')

    # Write to parquet (fast, columnar)
    df.to_parquet(parquet_path, index=False)

    return df