在pyspark中将Kafka的结果写入csv

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, StringType

spark = SparkSession.builder.appName("KafkaToCSV").getOrCreate()

kafka_topic = "your_kafka_topic"
kafka_bootstrap_servers = "your_kafka_bootstrap_servers"
kafka_group_id = "your_kafka_group_id"

df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", kafka_topic) \
    .option("startingOffsets", "latest") \
    .option("group.id", kafka_group_id) \
    .load()

schema = StructType([StructField("key", StringType(), True),
                     StructField("value", StringType(), True)])

df = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
    .select(from_json("value", schema).alias("data")) \
    .select("data.*")

output_path = "your_output_path"

query = df.writeStream.format("csv") \
    .option("path", output_path) \
    .option("checkpointLocation", "your_checkpoint_location") \
    .start()

query.awaitTermination()