如何基于PySpark/Pandas按用户ID补全60天内缺失日期？

我来帮你解决这个按用户补全日期的问题！你之前用Pandas重索引后其他字段变Null，核心原因是没按userId分组处理——咱们得针对每个用户单独生成完整的日期序列，再把原有数据匹配上去。下面给你三种不同技术栈的解决方案，按需选用：

适合小数据集场景，逻辑直观易调试：

首先构造你的测试数据，然后按用户分组生成日期序列：

import pandas as pd

# 构造原始数据
data = [
    ("2018-03-29", 55, "Large"),
    ("2018-03-30", 55, "small"),
    ("2018-03-29", 55, "x-small"),
    ("2018-04-20", 65, "Large"),
    ("2018-04-29", 75, "x-small")
]
df = pd.DataFrame(data, columns=["date", "userId", "classification"])
df["date"] = pd.to_datetime(df["date"])

def fill_user_dates(group):
    # 以用户最早记录日期为起点，生成60天的完整日期序列
    start_date = group["date"].min()
    date_range = pd.date_range(start=start_date, periods=60, freq="D")
    # 生成该用户的全日期基础表
    full_dates = pd.DataFrame({"date": date_range, "userId": group["userId"].iloc[0]})
    # 左连接原始数据，保留所有日期并匹配已有分类
    return full_dates.merge(group, on=["date", "userId"], how="left")

# 按用户分组处理后合并结果
filled_df = df.groupby("userId").apply(fill_user_dates).reset_index(drop=True)

# 可选：填充缺失的classification，比如用前一个有效值或默认值
filled_df["classification"] = filled_df.groupby("userId")["classification"].ffill().fillna("Unknown")

print(filled_df.head(10))

适合大数据量场景，分布式处理效率更高：

from pyspark.sql import SparkSession
from pyspark.sql.functions import min, sequence, explode, lit, date_add
from pyspark.sql.types import DateType

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

# 构造测试数据
data = [
    ("2018-03-29", 55, "Large"),
    ("2018-03-30", 55, "small"),
    ("2018-03-29", 55, "x-small"),
    ("2018-04-20", 65, "Large"),
    ("2018-04-29", 75, "x-small")
]
df = spark.createDataFrame(data, ["date", "userId", "classification"])
df = df.withColumn("date", df["date"].cast(DateType()))

# 计算每个用户的最早记录日期
user_start_dates = df.groupBy("userId").agg(min("date").alias("start_date"))

# 生成每个用户的60天日期序列并展开为行
full_dates_df = user_start_dates.withColumn(
    "date",
    explode(sequence("start_date", date_add("start_date", 59), lit(1)))
).drop("start_date")

# 左连接原始数据，补全分类信息
filled_df = full_dates_df.join(df, on=["userId", "date"], how="left")

# 可选：填充缺失值
filled_df = filled_df.fillna({"classification": "Unknown"})

filled_df.show(10)
spark.stop()

如果你的技术栈偏向Java，可以用Spark Java API实现：

import org.apache.spark.sql.*;
import org.apache.spark.sql.functions.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

public class FillDatesByUserId {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("FillDatesByUserId")
                .master("local[*]") // 本地测试用，生产环境移除
                .getOrCreate();

        // 构造测试数据
        Row[] data = {
                RowFactory.create("2018-03-29", 55, "Large"),
                RowFactory.create("2018-03-30", 55, "small"),
                RowFactory.create("2018-03-29", 55, "x-small"),
                RowFactory.create("2018-04-20", 65, "Large"),
                RowFactory.create("2018-04-29", 75, "x-small")
        };

        StructType schema = new StructType()
                .add("date", DataTypes.StringType)
                .add("userId", DataTypes.IntegerType)
                .add("classification", DataTypes.StringType);

        Dataset<Row> df = spark.createDataFrame(java.util.Arrays.asList(data), schema);
        // 转换日期列为Date类型
        df = df.withColumn("date", col("date").cast(DataTypes.DateType()));

        // 计算每个用户的最早日期
        Dataset<Row> userStartDates = df.groupBy("userId")
                .agg(min(col("date")).alias("start_date"));

        // 生成60天日期序列并展开
        Dataset<Row> fullDatesDF = userStartDates.withColumn(
                "date",
                explode(sequence(col("start_date"), date_add(col("start_date"), 59), lit(1)))
        ).drop("start_date");

        // 左连接原始数据
        Dataset<Row> filledDF = fullDatesDF.join(df,
                fullDatesDF.col("userId").equalTo(df.col("userId"))
                        .and(fullDatesDF.col("date").equalTo(df.col("date"))),
                "left"
        ).select(fullDatesDF.col("userId"), fullDatesDF.col("date"), df.col("classification"));

        // 填充缺失值
        filledDF = filledDF.fillna("Unknown", new String[]{"classification"});

        filledDF.show(10);
        spark.stop();
    }
}

额外说明 #

如果你的「60天时间范围」不是用户最早日期起算，而是其他规则（比如用户所有记录的日期区间、固定窗口等），只需要调整生成日期序列的逻辑即可。另外，若同一天存在多个classification记录，连接后会保留所有行，可根据需求用去重、合并数组等方式处理。

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