在不停止进程的情况下刷新Spark实时流中的数据帧

from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession

# 创建StreamingContext对象，设置批处理间隔时间为1秒
ssc = StreamingContext(sparkContext, 1)

# 使用StreamingContext对象创建一个DStream，表示实时流数据
dstream = ssc.socketTextStream("localhost", 9999)

# 对DStream进行转换操作，例如过滤、映射、聚合等
transformed_dstream = dstream.filter(lambda line: line.startswith("important"))

# 对每个RDD进行操作
transformed_dstream.foreachRDD(lambda rdd: 
    # 创建SparkSession对象
    spark = SparkSession.builder.getOrCreate()
    
    # 将RDD转换为数据帧
    df = spark.createDataFrame(rdd, schema)
    
    # 刷新数据帧，例如将数据帧写入外部存储系统
    df.write.format("parquet").save("hdfs://path/to/save")

    # 关闭SparkSession对象
    spark.stop()
)

# 启动StreamingContext
ssc.start()
ssc.awaitTermination()