按时间序列数据中的时间点在pyspark配置单元表的列中查找新值

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

# 初始化SparkSession
spark = SparkSession.builder.appName("TimeSeriesLookup").getOrCreate()

# 示例时间序列数据
time_series_data = [
    (1, "2023-01-01 10:00:00"),
    (2, "2023-01-01 11:00:00"),
    (3, "2023-01-01 12:00:00")
]
time_series_df = spark.createDataFrame(time_series_data, ["id", "timestamp"])

# 示例配置单元表
config_unit_table = [
    (1, "2023-01-01 09:00:00", "config_A"),
    (2, "2023-01-01 11:00:00", "config_B"),
    (3, "2023-01-01 13:00:00", "config_C")
]
config_unit_df = spark.createDataFrame(config_unit_table, ["id", "change_time", "config_value"])

# 将时间戳列转换为timestamp类型
time_series_df = time_series_df.withColumn("timestamp", col("timestamp").cast("timestamp"))
config_unit_df = config_unit_df.withColumn("change_time", col("change_time").cast("timestamp"))

# 使用窗口函数查找最近的配置值
from pyspark.sql.window import Window
window_spec = Window.orderBy("change_time").rowsBetween(Window.unboundedPreceding, Window.currentRow)

config_unit_df = config_unit_df.withColumn("latest_config", col("config_value"))

# 进行左连接查找新值
result_df = time_series_df.join(config_unit_df, on=col("timestamp") >= col("change_time"), how="left") \
    .select(time_series_df["id"], time_series_df["timestamp"], col("latest_config").alias("new_config"))

result_df.show(truncate=False)

time_series_df = time_series_df.withColumn("timestamp", to_timestamp(col("timestamp"), "yyyy-MM-dd HH:mm:ss"))