Aggregation and Grouping Questions

# read batch/deltas (deduped by op_id)
deltas = spark.read.stream(...)

# aggregate deltas per key in this micro-batch
batch_agg = deltas.groupBy("group_key").agg(
  sum("delta_value").alias("delta_sum"),
  max(struct("seq_no","event_ts","op_id")).alias("latest_meta"),
  max("is_delete").alias("has_delete")
)

# MERGE into materialized aggregate (Delta Lake)
from delta.tables import DeltaTable
target = DeltaTable.forName(spark, "mv_table")

target.alias("t").merge(
  batch_agg.alias("s"),
  "t.group_key = s.group_key"
).whenMatchedUpdate(
  condition = "s.latest_meta.seq_no > t.last_seq_no OR s.latest_meta.op_id > t.last_op_id",
  set = {
    "agg_sum": "t.agg_sum + s.delta_sum",
    "agg_count": "t.agg_count + s.delta_count",
    "last_seq_no": "s.latest_meta.seq_no",
    "last_op_id": "s.latest_meta.op_id",
    "tombstone_flag": "s.has_delete"
  }
).whenNotMatchedInsertAll().execute()

WITH customer_rev AS (
  -- total revenue per customer per state
  SELECT
    state,
    customer_id,
    SUM(amount) AS customer_revenue
  FROM sales
  GROUP BY state, customer_id
),
state_totals AS (
  -- total revenue per state
  SELECT
    state,
    SUM(customer_revenue) AS state_revenue
  FROM customer_rev
  GROUP BY state
),
ranked AS (
  -- rank customers within each state by revenue (ties handled with dense_rank)
  SELECT
    c.state,
    c.customer_id,
    c.customer_revenue,
    s.state_revenue,
    DENSE_RANK() OVER (PARTITION BY c.state ORDER BY c.customer_revenue DESC) AS rk
  FROM customer_rev c
  JOIN state_totals s USING (state)
)
SELECT
  state,
  customer_id AS top_customer_id,
  customer_revenue AS top_customer_revenue,
  state_revenue,
  ROUND(100.0 * customer_revenue / NULLIF(state_revenue,0), 2) AS pct_of_state_revenue
FROM ranked
WHERE rk = 1
ORDER BY state;

env
  .fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofHours(24)), "kafka")
  .assignTimestampsAndWatermarks(...)
  .keyBy(event -> event.userId)
  .process(new PerUserDeduper()) // emits (day, +1) only first time per user/day
  .keyBy(record -> record.day)
  .window(TumblingEventTimeWindows.of(Time.days(1)))
  .allowedLateness(Time.hours(24))
  .reduce(sumReducer)
  .sinkTo(kafkaUpsertSink); // transactional

SELECT
  customer_id,
  MIN(order_date) AS first_order_date,
  MAX(order_date) AS last_order_date
FROM orders
GROUP BY customer_id;

WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date ASC, order_id ASC) AS rn_first,
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC, order_id DESC) AS rn_last
  FROM orders
)
SELECT * FROM ranked WHERE rn_first = 1;  -- first order rows

-- or for last orders:
SELECT * FROM ranked WHERE rn_last = 1;

SELECT o.*
FROM orders o
JOIN (
  SELECT customer_id, MIN(order_date) AS first_dt
  FROM orders
  GROUP BY customer_id
) m ON o.customer_id = m.customer_id AND o.order_date = m.first_dt;

-- Count sessions where a product appears at least once
SELECT p.product_id, COUNT(DISTINCT s.session_id) AS sessions_count
FROM sessions s
-- ignore NULL products
CROSS JOIN LATERAL (
  SELECT jsonb_array_elements_text(s.products) AS product_id
) p
WHERE s.products IS NOT NULL
  AND jsonb_typeof(s.products) = 'array'
  AND (s.products <> '[]'::jsonb)
GROUP BY p.product_id
ORDER BY sessions_count DESC;

SELECT product_id, COUNT(DISTINCT session_id) AS sessions_count
FROM sessions,
UNNEST(
  CASE
    WHEN products IS NULL THEN [] -- treat NULL as empty
    WHEN JSON_TYPE(products) != 'ARRAY' THEN [] 
    ELSE JSON_EXTRACT_ARRAY(products)
  END
) AS prod
CROSS JOIN UNNEST([JSON_VALUE(prod)]) AS product_id
GROUP BY product_id
ORDER BY sessions_count DESC;