PySpark:如何在排序数据库中为下一个和前一个事件添加列？

from pyspark.sql import SparkSession

# 创建SparkSession对象
spark = SparkSession.builder \
    .appName("SortDB") \
    .getOrCreate()

# 从数据库加载数据集
df = spark.read.format("jdbc") \
    .option("url", "jdbc:mysql://localhost:3306/sortdb") \
    .option("dbtable", "events") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

from pyspark.sql.window import Window
from pyspark.sql.functions import lag, lead

# 定义窗口规范
windowSpec = Window.orderBy("event_time")

# 添加前一个事件列
df = df.withColumn("previous_event", lag(df.event, 1).over(windowSpec))

# 添加下一个事件列
df = df.withColumn("next_event", lead(df.event, 1).over(windowSpec))

# 将结果保存回数据库
df.write.format("jdbc") \
    .option("url", "jdbc:mysql://localhost:3306/sortdb") \
    .option("dbtable", "events_with_previous_next") \
    .option("user", "username") \
    .option("password", "password") \
    .mode("overwrite") \
    .save()