如何将spark数据帧中的多列写入kafka队列

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType
from pyspark.streaming import StreamingContext
from kafka import KafkaProducer

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

kafka_bootstrap_servers = "kafka服务器地址:端口号"
kafka_topic = "要写入的Kafka主题"

kafka_producer = KafkaProducer(bootstrap_servers=kafka_bootstrap_servers)

def write_to_kafka(row):
    kafka_producer.send(kafka_topic, str(row.asDict()).encode('utf-8'))

data_frame = spark.read.format("csv").option("header", "true").load("数据文件路径")

selected_columns = ["列1", "列2", "列3"]
selected_data_frame = data_frame.select(*selected_columns)

streaming_data = selected_data_frame.writeStream.foreach(write_to_kafka).start()

spark_streaming_context = StreamingContext(spark.sparkContext, 1)
spark_streaming_context.start()
spark_streaming_context.awaitTermination()