如何在Scala Spark中为DataFrame补充缺失的时间维度行？

原始DataFrame： #

+-------------+----------+------+
| key         | Time     | Value|
+-------------+----------+------+
|  1          |        1 | 1    | 
|  1          |        2 | 2    |
|  1          |        4 | 3    |
|  2          |        2 | 4    |
|  2          |        3 | 5    | 
+-------------+----------+------+

目标DataFrame： #

+-------------+----------+------+
| key         | Time     | Value|  
+-------------+----------+------+
|  1          |        1 | 1    | 
|  1          |        2 | 2    | 
|  1          |        3 | null | 
|  1          |        4 | 3    | 
|  2          |        1 | null |
|  2          |        2 | 4    |
|  2          |        3 | 5    | 
|  2          |        4 | null | 
+-------------+----------+------+

下面是几种可行的实现方法：

方法一：生成笛卡尔积后左连接 #

先提取所有唯一key和完整时间序列，生成两者的全组合（笛卡尔积），再通过左连接匹配原始数据的Value，缺失位置自动填充NaN（对应目标中的null）。

代码示例：

import pandas as pd

# 构造原始数据
df = pd.DataFrame({
    'key': [1,1,1,2,2],
    'Time': [1,2,4,2,3],
    'Value': [1,2,3,4,5]
})

# 定义完整时间点集合
full_times = [1,2,3,4]
# 获取所有唯一key
unique_keys = df['key'].unique()

# 生成key与时间点的全组合
full_combinations = pd.MultiIndex.from_product([unique_keys, full_times], names=['key', 'Time']).to_frame(index=False)
# 左连接原始数据补全Value
result = pd.merge(full_combinations, df, on=['key', 'Time'], how='left')

print(result)

方法二：分组后重新索引 #

按key分组，对每个分组用完整时间序列重新索引，自动补全缺失的时间点，Value列空缺处填充NaN。

代码示例：

import pandas as pd

df = pd.DataFrame({
    'key': [1,1,1,2,2],
    'Time': [1,2,4,2,3],
    'Value': [1,2,3,4,5]
})

full_times = [1,2,3,4]

# 分组后重新索引并合并结果
result = df.groupby('key').apply(
    lambda group: group.set_index('Time').reindex(full_times).reset_index()
).droplevel(0).reset_index(drop=True)

print(result)

方法三：透视表重塑后转长格式 #

先将原始数据转为透视表（行是key，列是Time），缺失的时间点会自动填充NaN，再将透视表转回长格式即可得到目标结构。

代码示例：

import pandas as pd

df = pd.DataFrame({
    'key': [1,1,1,2,2],
    'Time': [1,2,4,2,3],
    'Value': [1,2,3,4,5]
})

full_times = [1,2,3,4]

# 生成包含全时间点的透视表
pivot_df = df.pivot(index='key', columns='Time', values='Value').reindex(columns=full_times)
# 转回长格式并整理列名
result = pivot_df.unstack().reset_index(name='Value').rename(columns={'level_0': 'Time'}).sort_values(['key', 'Time']).reset_index(drop=True)

print(result)

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