本文介绍如何在 pyspark 中对结构一致的两个年份 dataframe（如 2020 和 2030）执行高效、可扩展的线性插值，生成中间年份（如 2025）的完整数值序列，避免依赖 pandas api on spark 或本地计算。
本文介绍如何在 pyspark 中对结构一致的两个年份 dataframe（如 2020 和 2030）执行高效、可扩展的线性插值，生成中间年份（如 2025）的完整数值序列，避免依赖 pandas api on spark 或本地计算。
在大规模时序数据处理场景中，常需基于已知年份（如基准年与目标年）的观测值，推算中间年份的合理估计值。PySpark 原生不提供 interpolate() 方法（该方法仅存在于 Pandas API on Spark，而非核心 PySpark SQL），因此直接迁移 Pandas 的 .interpolate() 逻辑会导致运行失败或性能瓶颈。正确做法是利用 PySpark 的列式计算能力，通过代数公式实现分布式线性插值。
线性插值的核心公式为：
[
v(t) = v_1 + (v_2 - v_1) \times \frac{t - t_1}{t_2 - t_1}
]
其中 (t_1, t_2) 为起止年份，(v_1, v_2) 为对应列的标量值，(t) 为待插值年份。
以下为完整、健壮的实现方案：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit
from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType
def interpolate_between_years(first: DataFrame, second: DataFrame, year_col: str = "year") -> DataFrame:
# 提取起止年份（确保单行且 year 列存在）
year_first = first.select(year_col).first()[0]
year_second = second.select(year_col).first()[0]
if year_first >= year_second:
raise ValueError("First year must be strictly less than second year.")
# 构建目标年份序列 DataFrame（含 year 列）
years_list = [(y,) for y in range(year_first, year_second + 1)]
schema = StructType([StructField(year_col, IntegerType(), True)])
interpolated_df = spark.createDataFrame(years_list, schema)
# 对每个非年份列执行向量化插值
for col_name in first.columns:
if col_name == year_col:
continue
# 获取首尾年份对应列的标量值（假设每 DataFrame 仅一行有效数据；若多行，需明确聚合策略，如 .agg(F.first())）
first_val = first.select(col_name).first()[0]
second_val = second.select(col_name).first()[0]
# 计算插值表达式：v1 + (v2 - v1) * (year - y1) / (y2 - y1)
diff = second_val - first_val
delta_year = year_second - year_first
interpolated_df = interpolated_df.withColumn(
col_name,
lit(first_val) + lit(diff) * (col(year_col) - lit(year_first)) / lit(delta_year)
)
return interpolated_df
# 示例使用
spark = SparkSession.builder.appName("LinearInterpolation").getOrCreate()
# 注意：示例中每个 DataFrame 应代表某一年份的「汇总快照」，即单行（或多行但需提前聚合）
df_2020 = spark.createDataFrame([(2020, 10.0, 20.0)], ["year", "value1", "value2"])
df_2030 = spark.createDataFrame([(2030, 40.0, 50.0)], ["year", "value1", "value2"])
result = interpolate_between_years(df_2020, df_2030)
result.orderBy("year").show(20, truncate=False)
✅ 关键说明与注意事项：
- 输入约束：本函数默认 first 和 second 各代表一个年份的聚合快照（单行）。若原始 DataFrame 包含多行（如按区域分组），请先使用 .groupBy(...).agg(...) 聚合至单行/键值对，再传入函数。
- 列类型安全：示例中显式指定 DoubleType，实际使用时建议检查数值列是否为数值类型（IntegerType/DoubleType），避免隐式转换错误。
- 性能优势：全程基于 Catalyst 优化器执行，所有计算在集群端完成，无需 collect() 或 toPandas()，适用于 TB 级数据。
- 扩展性：支持任意数量的数值列；只需确保两 DataFrame 列名、顺序完全一致（除 year 外）。
- 替代方案辨析：Pandas API on Spark 的 .interpolate() 虽语法相似，但底层仍转为 Pandas 执行，无法真正分布式，且不支持跨 DataFrame 插值逻辑——故不推荐用于生产环境的大规模任务。
最终输出 DataFrame 可直接用于后续分析，例如筛选特定年份：
result.filter(col("year") == 2025).select("value1", "value2").show()
# 输出：+------+------+
# |value1|value2|
# +------+------+
# | 25.0| 35.0|
# +------+------+
该方案兼顾准确性、可读性与工程可靠性，是 PySpark 生产环境中时间序列插值的标准实践。
