如何高效识别历史表中状态两次变更的SKU ID并生成透视表？

我有一张包含id、date、status字段的历史表，需要完成两个核心任务：

• 识别所有从状态1（Active，活跃）变更为0（Inactive，非活跃）后又恢复为1（Active）的SKU ID。
• 生成以所有唯一SKU ID为行、日期为列的透视表，单元格对应SKU在对应日期的状态值。

由于历史表包含约1000万唯一SKU ID，现寻求高效的数据处理方案。

数据样例 #

id       date       status
I01234   5/12/2023  1
I14690   4/13/2021  0

期望最终数据结构 #

id      01/01/2021 01/02/2021 01/03/2021 ........ 10/31/2023
I01234  1          1          1                   0
I14690  1          0          0                   1

一、识别符合状态变更条件的SKU ID #

针对千万级数据，必须依赖数据库原生窗口函数实现高效计算，避免重复全表扫描。核心思路是标记每个SKU的状态变更节点，再筛选同时存在「1→0」和「0→1」变更的SKU。

示例SQL（以MySQL为例，其他数据库语法略有差异） #

-- 步骤1：标记每个SKU的状态变更记录
WITH status_changes AS (
    SELECT
        id,
        date,
        status,
        -- 获取上一条记录的状态
        LAG(status) OVER (PARTITION BY id ORDER BY date) AS prev_status,
        -- 标记从活跃到非活跃的变更
        CASE WHEN prev_status = 1 AND status = 0 THEN 1 ELSE 0 END AS is_active_to_inactive,
        -- 标记从非活跃到活跃的变更
        CASE WHEN prev_status = 0 AND status = 1 THEN 1 ELSE 0 END AS is_inactive_to_active
    FROM your_history_table
),
-- 步骤2：统计每个SKU是否存在两种目标变更
sku_change_summary AS (
    SELECT
        id,
        MAX(is_active_to_inactive) AS has_active_inactive,
        MAX(is_inactive_to_active) AS has_inactive_active
    FROM status_changes
    GROUP BY id
)
-- 步骤3：筛选符合条件的SKU
SELECT id
FROM sku_change_summary
WHERE has_active_inactive = 1 AND has_inactive_active = 1;

关键优化点：给id和date创建联合索引，能大幅提升窗口函数的执行效率，避免全表扫描。

二、生成大维度透视表 #

直接生成包含1000万行+上千列的宽表会面临内存溢出、存储爆炸的问题，需针对性优化：

方案1：缩小数据范围后生成透视表 #

• 只针对第一步筛选出的目标SKU生成透视表，将行数从1000万压缩到符合条件的子集。
• 按日期范围分片生成（比如按月份拆分），再按需合并结果，降低单次计算的资源压力。

方案2：用分布式计算框架处理（推荐） #

针对超大规模数据，使用Spark等分布式框架利用集群资源并行计算，避免单机瓶颈：

from pyspark.sql import SparkSession
from pyspark.sql.functions import first

# 初始化Spark会话
spark = SparkSession.builder.appName("SKUPivot").getOrCreate()

# 读取历史表数据
df = spark.read.table("your_history_table")

# 生成透视表：以id为行，date为列，取当日第一条状态记录
pivot_df = df.groupBy("id").pivot("date").agg(first("status"))

# 可选：仅保留符合状态变更条件的SKU
target_skus = spark.read.table("target_sku_table")  # 第一步得到的SKU列表
final_pivot_df = pivot_df.join(target_skus, on="id", how="inner")

# 写入列式存储（如Parquet），方便后续高效查询
final_pivot_df.write.parquet("sku_pivot_result")

方案3：维护每日快照表优化透视 #

如果历史表是增量更新的，可预先维护每日快照表：

• 每天计算每个SKU的当日最新状态，存储到快照表（结构：id, date, status），数据量仅为「SKU数×天数」，远小于原始历史表。
• 基于快照表生成透视表，计算效率会大幅提升。

核心注意事项：

• 绝对避免用Excel、单机Pandas等工具直接处理1000万级数据，会直接触发内存溢出。
• 优先利用数据库索引、分布式计算的并行能力，减少不必要的数据扫描。

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