3-4. 重複排除と集約(count・approx_count_distinct・mean・summary)

💡 具体例:全列一致の重複と、キー列だけの重複

df = spark.read.table("bronze.orders_raw")

# (1) 全列が完全に一致する行を1行に(二重取り込みの掃除)
df_unique = df.dropDuplicates()          # df.distinct() と同じ

# (2) order_id が同じなら他の列が違っても1行に
df_unique = df.dropDuplicates(["order_id"])

(2)には重要な注意点があります。dropDuplicates(["order_id"]) は同じ order_id の行のうちどの1行が残るかを保証しません。「最新の1行を残したい」という要件には、次のウィンドウ関数を使います。

💡 具体例:order_id ごとに最新レコードを抽出する

from pyspark.sql import functions as F
from pyspark.sql.window import Window

w = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())

df_latest = (df
    .withColumn("rn", F.row_number().over(w))
    .filter(F.col("rn") == 1)
    .drop("rn")
)

-- SQL 版
SELECT * EXCEPT (rn)
FROM (
  SELECT *,
         row_number() OVER (
           PARTITION BY order_id
           ORDER BY updated_at DESC
         ) AS rn
  FROM bronze.orders_raw
)
WHERE rn = 1;

partitionBy が「どの単位で番号を振り直すか(キー)」、orderBy ... desc() が「何を1番とするか(最新)」を決めます。この形は 3-1 の MERGE 前のソース一意化にもそのまま使えます。

💡 具体例:地域×月の売上サマリー

from pyspark.sql import functions as F

summary = (spark.read.table("silver.orders")
    .withColumn("order_month", F.date_trunc("month", "order_date"))
    .groupBy("region", "order_month")
    .agg(
        F.count("*").alias("order_cnt"),                    # 行数
        F.countDistinct("customer_id").alias("customers"),  # 一意顧客数(正確)
        F.sum("total_amount").alias("revenue"),
        F.avg("total_amount").alias("avg_amount"),          # mean でも同じ
        F.min("order_date").alias("first_order"),
        F.max("order_date").alias("last_order"),
    )
)

-- SQL 版
SELECT
  region,
  date_trunc('month', order_date)  AS order_month,
  COUNT(*)                         AS order_cnt,
  COUNT(DISTINCT customer_id)      AS customers,
  SUM(total_amount)                AS revenue,
  AVG(total_amount)                AS avg_amount,
  MIN(order_date)                  AS first_order,
  MAX(order_date)                  AS last_order
FROM silver.orders
GROUP BY region, date_trunc('month', order_date);

mean と avg は同じ関数の別名です。また count("列名") はその列が null でない行だけを数え、count("*") は null を含む全行を数える、という違いにも注意してください。