Aggregation Functions and Group By Questions

-- split on non-word chars, lower-case, filter short tokens and common stopwords_table
WITH tokens AS (
  SELECT r.category, lower(regexp_split_to_table(review_text, '\W+')) AS token
  FROM reviews r
),
filtered AS (
  SELECT category, token
  FROM tokens
  WHERE length(token) > 2
    AND token NOT IN (SELECT word FROM stopwords)
)
SELECT category, token, count(*) AS freq
FROM filtered
GROUP BY category, token
ORDER BY category, freq DESC;
-- limit to top 10 per category depending on DB: use ROW_NUMBER() OVER (PARTITION BY category ORDER BY freq DESC)

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, lower, regexp_replace
from pyspark.ml.feature import Tokenizer, StopWordsRemover, CountVectorizer

spark = SparkSession.builder.getOrCreate()
df = spark.read.parquet("s3://.../reviews/")
# simple cleaning
df = df.withColumn("text_clean", lower(regexp_replace("review_text", r'[^A-Za-z0-9\s]', ' ')))
tokenizer = Tokenizer(inputCol="text_clean", outputCol="tokens")
tokens = tokenizer.transform(df)
remover = StopWordsRemover(inputCol="tokens", outputCol="filtered")
filtered = remover.transform(tokens)
# explode and count per category
exploded = filtered.select("category", explode("filtered").alias("token"))
top10 = (exploded.groupBy("category","token").count()
         .withColumn("rn", row_number().over(Window.partitionBy("category").orderBy(desc("count"))))
         .filter("rn <= 10"))

SELECT
  product_id,
  COUNT(*)        AS reviews_last_365,
  AVG(rating)     AS avg_rating_last_365
FROM reviews
WHERE
  review_date >= CURRENT_DATE - INTERVAL '365 days'  -- last 365 days
  AND rating IS NOT NULL                              -- exclude NULL ratings
GROUP BY product_id
HAVING
  COUNT(*) >= 50         -- at least 50 reviews
  AND AVG(rating) >= 4;  -- average rating >= 4

-- Dashboard A: using events_a table
SELECT DATE(event_timestamp AT TIME ZONE 'UTC') AS day_utc,
       COUNT(DISTINCT user_id) AS dau_a
FROM events_a
WHERE event_timestamp >= '2025-11-01' AND event_timestamp < '2025-11-08'
GROUP BY 1
ORDER BY 1;

-- Dashboard B: using events_b table or materialized view
SELECT DATE(event_time AT TIME ZONE 'America/Los_Angeles') AS day_pst,
       COUNT(DISTINCT user_id) AS dau_b
FROM events_b
WHERE event_time >= '2025-11-01' AND event_time < '2025-11-08'
GROUP BY 1
ORDER BY 1;

SELECT DATE((event_timestamp AT TIME ZONE 'UTC')) AS day_utc,
       COUNT(DISTINCT user_id) FROM events_a GROUP BY 1;
SELECT DATE((event_time AT TIME ZONE 'UTC')) AS day_utc,
       COUNT(DISTINCT user_id) FROM events_b GROUP BY 1;

-- Are dashboards using DISTINCT user_id or counting events?
SELECT day_utc, COUNT(*) AS events, COUNT(DISTINCT user_id) AS distinct_users
FROM (
  SELECT user_id, DATE(event_timestamp AT TIME ZONE 'UTC') AS day_utc FROM events_a
) t GROUP BY day_utc;

-- Identify duplicate event_ids or identical rows
SELECT event_id, COUNT(*) FROM events_a GROUP BY event_id HAVING COUNT(*)>1 LIMIT 10;

-- Or duplicates per user-day
SELECT user_id, day_utc, COUNT(*) AS hits
FROM (
  SELECT user_id, DATE(event_timestamp AT TIME ZONE 'UTC') AS day_utc, event_signature
  FROM events_a
) t
GROUP BY user_id, day_utc HAVING COUNT(*) > 1 ORDER BY hits DESC LIMIT 20;

-- Compare predicates used
SELECT COUNT(DISTINCT user_id) FROM events_a WHERE event_type IN ('open','engage');
SELECT COUNT(DISTINCT user_id) FROM events_a WHERE user_agent NOT LIKE '%bot%';

-- Users counted in A but not B for a given day
WITH a AS (
  SELECT DISTINCT user_id FROM events_a WHERE DATE(event_timestamp AT TIME ZONE 'UTC')='2025-11-05'
),
b AS (
  SELECT DISTINCT user_id FROM events_b WHERE DATE(event_time AT TIME ZONE 'UTC')='2025-11-05'
)
SELECT 'only_in_a' AS side, a.user_id FROM a LEFT JOIN b USING (user_id) WHERE b.user_id IS NULL
UNION ALL
SELECT 'only_in_b', b.user_id FROM b LEFT JOIN a USING (user_id) WHERE a.user_id IS NULL;

-- Flatten payload.items, extract item_id and numeric price, sum revenue per item_id
WITH items AS (
  SELECT
    e.id AS event_id,
    elem ->> 'item_id' AS item_id_text,
    -- extract price as text then cast to numeric safely
    NULLIF(elem ->> 'price', '') AS price_text
  FROM events e
  CROSS JOIN LATERAL
    jsonb_array_elements(COALESCE(e.payload -> 'items', '[]'::jsonb)) AS elem
)
, parsed AS (
  SELECT
    event_id,
    item_id_text,
    -- only keep rows with non-null item_id
    CASE WHEN item_id_text IS NOT NULL THEN item_id_text END AS item_id,
    -- try to cast price_text to numeric; invalid casts become NULL
    CASE
      WHEN price_text ~ '^[0-9]+(\.[0-9]+)?$' THEN (price_text)::numeric
      ELSE NULL
    END AS price
  FROM items
)
SELECT
  item_id,
  SUM(price) AS total_revenue,
  COUNT(*) FILTER (WHERE price IS NOT NULL) AS price_count,
  COUNT(*) AS rows_seen
FROM parsed
WHERE item_id IS NOT NULL
GROUP BY item_id
ORDER BY total_revenue DESC;