在pyspark中将Kafka的结果写入csv

可以通过以下步骤实现：

导入所需的模块和库：

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

创建SparkSession对象：

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

定义Kafka主题和相关配置：

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

读取Kafka数据：

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()

解析Kafka数据：

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.*")

将结果写入csv文件：

output_path = "your_output_path"

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

query.awaitTermination()

在上述代码中，需要替换以下内容：