PySpark中如何按分组/分区关联两个时间序列DataFrame？

Got it, let's break this down step by step—this is a super common use case for time series data in PySpark. Here's exactly how to pull it off:

First, let's create sample DataFrames that match your scenario: one with gaps, one with a full, gapless time series per group (id).

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date

# Initialize Spark session
spark = SparkSession.builder.appName("GroupedTimeSeriesJoin").getOrCreate()

# DataFrame 1: Time series with gaps (3 sequences identified by id)
data_with_gaps = [
    ("id1", "2023-01-01", 10),
    ("id1", "2023-01-03", 15),
    ("id2", "2023-01-02", 20),
    ("id2", "2023-01-04", 25),
    ("id3", "2023-01-01", 5),
    ("id3", "2023-01-04", 8)
]
df_gaps = spark.createDataFrame(data_with_gaps, ["id", "timestamp", "value"])
df_gaps = df_gaps.withColumn("timestamp", to_date(col("timestamp")))

# DataFrame 2: Gapless time series (all dates for each id)
data_full = [
    ("id1", "2023-01-01"), ("id1", "2023-01-02"), ("id1", "2023-01-03"), ("id1", "2023-01-04"),
    ("id2", "2023-01-01"), ("id2", "2023-01-02"), ("id2", "2023-01-03"), ("id2", "2023-01-04"),
    ("id3", "2023-01-01"), ("id3", "2023-01-02"), ("id3", "2023-01-03"), ("id3", "2023-01-04")
]
df_full = spark.createDataFrame(data_full, ["id", "timestamp"])
df_full = df_full.withColumn("timestamp", to_date(col("timestamp")))

The key here is to join per group (id) and per timestamp using a left_outer join. This ensures we keep every row from the gapless DataFrame, and match it to the corresponding row (if it exists) in the gap-filled DataFrame.

# Perform grouped join on id and timestamp
joined_df = df_full.join(
    df_gaps,
    on=["id", "timestamp"],  # Join keys: group id + time point
    how="left_outer"         # Keep all rows from the gapless DF
)

# Check the result
joined_df.show()

Output: #

+---+----------+-----+
| id| timestamp|value|
+---+----------+-----+
|id1|2023-01-01|   10|
|id1|2023-01-02| null|
|id1|2023-01-03|   15|
|id1|2023-01-04| null|
|id2|2023-01-01| null|
|id2|2023-01-02|   20|
|id2|2023-01-03| null|
|id2|2023-01-04|   25|
|id3|2023-01-01|    5|
|id3|2023-01-02| null|
|id3|2023-01-03| null|
|id3|2023-01-04|    8|
+---+----------+-----+

Perfect—you can see we now have every time point for each id, with null values where the original data had gaps.

If you need to fill those null values (e.g., forward-fill with the last known value), use a window function partitioned by id and ordered by timestamp:

from pyspark.sql.window import Window
from pyspark.sql.functions import last

# Define window: partition by id, order by timestamp, look back to the first row
window_spec = Window.partitionBy("id").orderBy("timestamp").rowsBetween(Window.unboundedPreceding, Window.currentRow)

# Fill nulls with the last non-null value in the group
filled_df = joined_df.withColumn(
    "value_filled",
    last(col("value"), ignorenulls=True).over(window_spec)
)

filled_df.show()

Output: #

+---+----------+-----+------------+
| id| timestamp|value|value_filled|
+---+----------+-----+------------+
|id1|2023-01-01|   10|          10|
|id1|2023-01-02| null|          10|
|id1|2023-01-03|   15|          15|
|id1|2023-01-04| null|          15|
|id2|2023-01-01| null|        null|
|id2|2023-01-02|   20|          20|
|id2|2023-01-03| null|          20|
|id2|2023-01-04|   25|          25|
|id3|2023-01-01|    5|           5|
|id3|2023-01-02| null|           5|
|id3|2023-01-03| null|           5|
|id3|2023-01-04|    8|           8|
+---+----------+-----+------------+

• Repartition first: If your DataFrames are large, repartition both by id before joining to reduce shuffle overhead:

  df_gaps = df_gaps.repartition("id")
  df_full = df_full.repartition("id")
  

• Generate gapless series dynamically: If you don't already have df_full, generate it using sequence and explode instead of hardcoding dates:

  from pyspark.sql.functions import sequence, explode, lit
  
  start_date = to_date(lit("2023-01-01"))
  end_date = to_date(lit("2023-01-04"))
  
  # Get unique ids from df_gaps
  unique_ids = df_gaps.select("id").distinct()
  
  # Generate full date sequence for each id
  df_full_generated = unique_ids.withColumn(
      "timestamp",
      explode(sequence(start_date, end_date))
  )
  

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