PySpark :将Spark Dataframe写入Kafka主题

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# 创建SparkSession
spark = SparkSession.builder \
    .appName("PySpark Kafka Integration") \
    .getOrCreate()

# 读取数据到Spark Dataframe
data = spark.read.format("csv").option("header", "true").load("data.csv")

# 将数据写入Kafka主题
data.selectExpr("CAST(column1 AS STRING) AS key", "to_json(struct(*)) AS value") \
    .write \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka_server:9092") \
    .option("topic", "topic_name") \
    .save()

# 关闭SparkSession
spark.stop()