如何用SQL、Python或R合并缓慢变化维度表并生成历史宽表？

需求本质是将带有时间生效区间的多属性长表，转换为按时间切片的宽表——每个切片对应某一时间段内，指定ID的所有属性当前值，支持多ID和未知属性类型（type）。以下是SQL、Python、R三种工具的高效实现方式：

一、SQL实现 #

核心思路：先提取每个ID的所有时间断点（所有date_from和date_to），生成连续的时间区间；再将每个区间内的属性值匹配到对应区间，最后动态将属性类型转为列（适配未知type）。

PostgreSQL示例 #

-- 1. 生成每个ID的时间区间
WITH time_breaks AS (
    SELECT id, date_from AS break_time FROM your_table
    UNION
    SELECT id, date_to AS break_time FROM your_table
),
sorted_breaks AS (
    SELECT 
        id, 
        break_time,
        LEAD(break_time) OVER (PARTITION BY id ORDER BY break_time) AS next_break
    FROM time_breaks
    WHERE break_time <> '9999-12-31'
),
time_intervals AS (
    SELECT 
        id,
        break_time AS date_from,
        COALESCE(next_break, '9999-12-31') AS date_to
    FROM sorted_breaks
    WHERE next_break IS NOT NULL
)
-- 2. 动态构造PIVOT语句（适配未知type）
SELECT string_agg(DISTINCT quote_ident(type), ', ') INTO attr_cols FROM your_table;

EXECUTE format('
    SELECT 
        ti.id,
        %s,
        ti.date_from,
        ti.date_to
    FROM time_intervals ti
    LEFT JOIN your_table t ON ti.id = t.id 
        AND ti.date_from >= t.date_from 
        AND ti.date_to <= t.date_to
    GROUP BY ti.id, ti.date_from, ti.date_to
    PIVOT (
        MAX(t.value) FOR t.type IN (%s)
    ) AS pivoted
    ORDER BY ti.id, ti.date_from
', attr_cols, attr_cols);

MySQL示例（无原生PIVOT，用条件聚合） #

-- 1. 生成动态属性列的SQL片段
SELECT GROUP_CONCAT(DISTINCT CONCAT('MAX(CASE WHEN type = ''', type, ''' THEN value END) AS ', type)) INTO attr_sql FROM your_table;

-- 2. 构造完整查询
SET @sql = CONCAT('
    WITH time_breaks AS (
        SELECT id, date_from AS break_time FROM your_table
        UNION
        SELECT id, date_to AS break_time FROM your_table
    ),
    sorted_breaks AS (
        SELECT 
            id, 
            break_time,
            LEAD(break_time) OVER (PARTITION BY id ORDER BY break_time) AS next_break
        FROM time_breaks
        WHERE break_time <> ''9999-12-31''
    ),
    time_intervals AS (
        SELECT 
            id,
            break_time AS date_from,
            COALESCE(next_break, ''9999-12-31'') AS date_to
        FROM sorted_breaks
        WHERE next_break IS NOT NULL
    )
    SELECT 
        ti.id,
        ', attr_sql, ',
        ti.date_from,
        ti.date_to
    FROM time_intervals ti
    LEFT JOIN your_table t ON ti.id = t.id 
        AND ti.date_from >= t.date_from 
        AND ti.date_to <= t.date_to
    GROUP BY ti.id, ti.date_from, ti.date_to
    ORDER BY ti.id, ti.date_from
');

PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

二、Python（Pandas）实现 #

核心思路：按ID分组处理，生成每个ID的时间区间后，匹配区间内的有效属性记录，最后转成宽表。

import pandas as pd

# 读取并预处理数据
df = pd.read_csv("your_data.csv")
df["date_from"] = pd.to_datetime(df["date_from"])
df["date_to"] = pd.to_datetime(df["date_to"])

# 定义单ID处理函数
def process_single_id(group):
    # 提取时间断点并生成区间
    breaks = pd.concat([group["date_from"], group["date_to"]]).unique()
    breaks = sorted(breaks[breaks != pd.to_datetime("9999-12-31")])
    breaks.append(pd.to_datetime("9999-12-31"))
    
    intervals = []
    for i in range(len(breaks)-1):
        intervals.append({
            "id": group["id"].iloc[0],
            "date_from": breaks[i],
            "date_to": breaks[i+1]
        })
    intervals_df = pd.DataFrame(intervals)
    
    # 匹配区间内的有效属性
    merged = intervals_df.merge(group, on="id", how="left")
    valid_mask = (merged["date_from_x"] >= merged["date_from_y"]) & (merged["date_to_x"] <= merged["date_to_y"])
    valid_data = merged[valid_mask]
    
    # 转宽表并整理列名
    wide_df = valid_data.pivot_table(
        index=["id", "date_from_x", "date_to_x"],
        columns="type",
        values="value",
        aggfunc="first"
    ).reset_index()
    wide_df.rename(columns={"date_from_x": "date_from", "date_to_x": "date_to"}, inplace=True)
    
    # 尝试转换数值类型（如headcount）
    for col in wide_df.columns:
        if col not in ["id", "date_from", "date_to"]:
            try:
                wide_df[col] = pd.to_numeric(wide_df[col])
            except:
                pass
    return wide_df

# 分组处理并合并结果
final_result = df.groupby("id").apply(process_single_id).reset_index(drop=True)
print(final_result)

三、R实现 #

核心思路：用dplyr分组处理时间区间，tidyr转换宽表，逻辑与Python一致。

library(dplyr)
library(tidyr)
library(lubridate)

# 读取并预处理数据
df <- read.csv("your_data.csv") %>%
  mutate(across(c(date_from, date_to), ymd))

# 定义单ID处理函数
process_single_id <- function(group) {
    # 生成时间区间
    breaks <- c(group$date_from, group$date_to) %>%
      unique() %>%
      .[. != ymd("9999-12-31")] %>%
      sort() %>%
      c(., ymd("9999-12-31"))
    
    intervals <- tibble(
      id = group$id[1],
      date_from = breaks[-length(breaks)],
      date_to = breaks[-1]
    )
    
    # 匹配有效属性
    merged_data <- intervals %>%
      left_join(group, by = "id") %>%
      filter(date_from.x >= date_from.y, date_to.x <= date_to.y)
    
    # 转宽表并整理
    wide_df <- merged_data %>%
      pivot_wider(
        id_cols = c(id, date_from.x, date_to.x),
        names_from = type,
        values_from = value,
        values_fn = first
      ) %>%
      rename(date_from = date_from.x, date_to = date_to.x)
    
    # 尝试转换数值类型
    wide_df <- wide_df %>%
      mutate(across(-c(id, date_from, date_to), ~suppressWarnings(as.numeric(.))))
    return(wide_df)
}

# 分组处理并合并结果
final_result <- df %>%
  group_by(id) %>%
  group_modify(~process_single_id(.x)) %>%
  ungroup()

print(final_result)

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