Aggregation Functions and Group By Questions

-- BigQuery: exact distinct (may be expensive for huge user_id cardinality)
SELECT country, device, COUNT(DISTINCT user_id) AS distinct_users
FROM `project.dataset.events`
GROUP BY country, device;

SELECT
  SPLIT(k, '|')[OFFSET(0)] AS country,
  SPLIT(k, '|')[OFFSET(1)] AS device,
  COUNT(DISTINCT SPLIT(k, '|')[OFFSET(2)]) AS distinct_users
FROM (
  SELECT CONCAT(country, '|', device, '|', user_id) AS k
  FROM `project.dataset.events`
)
GROUP BY country, device;

SELECT
  country,
  device,
  APPROX_COUNT_DISTINCT(user_id) AS approx_distinct_users
FROM `project.dataset.events`
GROUP BY country, device;

-- create per-row HLL sketch
WITH sketches AS (
  SELECT
    country,
    device,
    HLL_COUNT.INIT(user_id) AS hll -- create HLL sketch per row/partition
  FROM `project.dataset.events`
)
-- aggregate/merge sketches per group then extract estimate
SELECT
  country,
  device,
  HLL_COUNT.EXTRACT(HLL_COUNT.MERGE_AGG(hll)) AS approx_distinct_users
FROM sketches
GROUP BY country, device;

-- Portable ANSI-style using EXTRACT for year/month
SELECT
  customer_id,
  EXTRACT(YEAR FROM created_at) AS year,
  EXTRACT(MONTH FROM created_at) AS month,
  -- optional month label yyyy-mm
  CAST(EXTRACT(YEAR FROM created_at) AS VARCHAR) || '-' ||
    LPAD(CAST(EXTRACT(MONTH FROM created_at) AS VARCHAR), 2, '0') AS month_label,
  SUM(total_amount) AS total_revenue
FROM orders
GROUP BY customer_id, EXTRACT(YEAR FROM created_at), EXTRACT(MONTH FROM created_at)
ORDER BY customer_id, year, month;

SELECT
  customer_id,
  DATE_TRUNC('month', created_at) AS month_start,
  SUM(total_amount) AS total_revenue
FROM orders
GROUP BY customer_id, DATE_TRUNC('month', created_at)
ORDER BY customer_id, month_start;

from pyspark.sql import functions as F
from pyspark.sql import types as T

# assume `orders` is the input DataFrame with columns:
# order_id, product_id, price, quantity, created_at (string or timestamp)

# 1) Light-weight projection + correct types
df = (orders
      .select(
         F.col("order_id"),
         F.col("product_id"),
         (F.col("price").cast(T.DecimalType(18,2))).alias("price"),
         F.col("quantity").cast("long").alias("quantity"),
         F.to_timestamp("created_at").alias("created_at")
      )
     )

# 2) Compute revenue and week bucket (ISO week or custom)
df = df.withColumn("revenue", F.col("price") * F.col("quantity")) \
       .withColumn("week_start", F.date_trunc("week", F.col("created_at")).cast("date"))

# 3) Optional: reduce shuffle by pre-partitioning before expensive groupBy if dataset is huge.
# If downstream writes are per product, repartition by product_id (or product_id + week_start)
df = df.repartition(200, F.col("product_id"))  # tune num partitions to cluster

# 4) Aggregation (Spark will do map-side partial aggregates)
result = (df.groupBy("product_id", "week_start")
            .agg(F.sum("revenue").alias("weekly_revenue"))
            .select("product_id", "week_start", F.round("weekly_revenue", 2).alias("weekly_revenue"))
         )

# 5) Write out partitioned by week for efficient reads
(result.write
       .mode("overwrite")
       .partitionBy("week_start")
       .parquet("/path/to/output/weekly_revenue_by_product"))

with raw as (
  select
    product_id,
    date_trunc('day', happened_at) as revenue_date,
    amount,
    updated_at
  from {{ ref('events_staging') }}
  {% if is_incremental() %}
    where happened_at >= (select min(revenue_date) from {{ this }}) - interval '7 day'
    -- and/or filter by updated_at for CDC: updated_at >= (select max(updated_at) from {{ this }}) - interval '1 day'
  {% endif %}
),

agg as (
  select
    product_id,
    revenue_date,
    sum(amount) as daily_revenue,
    max(updated_at) as last_updated
  from raw
  group by product_id, revenue_date
)

-- For warehouses that support MERGE (Snowflake/BigQuery/SQL Server):
merge into {{ this }} tgt
using agg src
on tgt.product_id = src.product_id and tgt.revenue_date = src.revenue_date
when matched and src.last_updated > tgt.last_updated then
  update set daily_revenue = src.daily_revenue, last_updated = src.last_updated
when not matched then
  insert (product_id, revenue_date, daily_revenue, last_updated)
  values (src.product_id, src.revenue_date, src.daily_revenue, src.last_updated);