如何用Snowflake SQL/Snowpark/Pandas筛选连续12个月有数据的ID

示例数据 #

需求说明 #

筛选出过去12个连续月份均有销售数据的id（即每个月都存在对应数据）。例如id=1每月都有销售记录，但id=2缺失部分月份的数据。

Snowflake SQL实现 #

先生成过去12个连续月份的月末日期，关联去重后的销售数据，统计每个id覆盖的月份数，筛选出覆盖全部12个月的id：

-- 生成过去12个连续月份的月末日期
WITH required_months AS (
    SELECT LAST_DAY(DATEADD(MONTH, -seq8(), DATE_TRUNC('MONTH', CURRENT_DATE)))::DATE AS monthend
    FROM TABLE(GENERATOR(ROWCOUNT => 12))
),
-- 去重同一id同一月份的销售记录，避免重复计数
unique_sales AS (
    SELECT DISTINCT id, sales_date_monthend
    FROM sample_table
)
SELECT id
FROM required_months
LEFT JOIN unique_sales ON required_months.monthend = unique_sales.sales_date_monthend
GROUP BY id
HAVING COUNT(unique_sales.sales_date_monthend) = 12;

Snowpark实现 #

利用Snowpark DataFrame API，构建全量id-月份组合，左连接销售数据后统计缺失月份数，筛选无缺失的id：

from snowflake.snowpark import Window
import snowflake.snowpark.functions as F

# 生成过去12个连续月份的月末日期
date_range = session.sql("""
    SELECT LAST_DAY(DATEADD(MONTH, -seq8(), DATE_TRUNC('MONTH', CURRENT_DATE)))::DATE AS monthend
    FROM TABLE(GENERATOR(ROWCOUNT => 12))
""")

# 读取原表并去重id和月份
sample_table = session.table("sample_table").select("id", "sales_date_monthend").distinct()

# 生成每个id对应所有12个月份的组合
all_id_months = sample_table.select("id").distinct().crossJoin(date_range)

# 左连接销售数据，统计每个id的缺失月份数
joined = all_id_months.join(
    sample_table, 
    (all_id_months["id"] == sample_table["id"]) & 
    (all_id_months["monthend"] == sample_table["sales_date_monthend"]),
    how="left"
)

result = joined.group_by("id") \
               .agg(F.count(F.when(F.col("sales_date_monthend").is_null(), 1)).alias("missing_months")) \
               .filter(F.col("missing_months") == 0) \
               .select("id")

# 输出结果
result.show()

Pandas实现 #

本地处理时，统一日期格式后，检查每个id的月份集合是否包含过去12个连续月份：

import pandas as pd
from datetime import datetime
from dateutil.relativedelta import relativedelta

# 读取示例数据（实际可从Snowflake导入）
data = [
    [1,5,"2023-05-31"],[1,10,"2023-04-30"],[1,11,"2023-03-31"],[1,12,"2023-02-27"],
    [1,1,"2023-01-31"],[1,5,"2022-12-31"],[1,3,"2022-11-30"],[1,6,"2022-10-31"],
    [1,9,"2022-09-30"],[1,8,"2022-08-31"],[1,7,"2022-07-31"],[1,3,"2023-06-30"],
    [1,2,"2023-05-31"],[2,8,"2022-08-31"],[2,7,"2022-07-31"],[2,3,"2023-06-30"],
    [2,2,"2023-05-31"]
]
df = pd.DataFrame(data, columns=["id", "sales", "sales_date_monthend"])

# 统一日期格式为月末标准格式
df["sales_date_monthend"] = pd.to_datetime(df["sales_date_monthend"]).dt.to_period('M').dt.to_timestamp('M')

# 生成过去12个连续月份的月末集合
current_month = datetime.now().replace(day=1)
required_months = {current_month - relativedelta(months=i) for i in range(12)}

# 筛选符合条件的id
valid_ids = []
for id_, group in df.groupby("id"):
    id_months = set(group["sales_date_monthend"])
    if required_months.issubset(id_months):
        valid_ids.append(id_)

print("符合条件的id:", valid_ids)

内容的提问来源于stack exchange，提问作者Dametime