spark sql加载kafka数据并将查询结果写出到kafka的执行过程

可以使用Spark SQL集成Kafka功能来将查询结果写入Kafka，具体过程为：

1. 创建KafkaProducer，并指定Topic名称、消息Key的序列化方式、消息Value的序列化方式；
2. 使用Spark SQL查询需要输出到Kafka的数据，并将结果收集到Driver端；
3. 遍历查询结果，将每条记录转换成Kafka ProducerRecord，并将其发送到Kafka Topic中。

示例代码如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_json, struct
from kafka import KafkaProducer

# 创建SparkSession
spark = SparkSession.builder.appName("Write to Kafka").getOrCreate()

# Kafka配置项
bootstrap_servers = "localhost:9092"
topic = "test"
key_serializer = "org.apache.kafka.common.serialization.StringSerializer"
value_serializer = "org.apache.kafka.common.serialization.StringSerializer"

# 创建KafkaProducer
producer = KafkaProducer(bootstrap_servers=bootstrap_servers,
                         key_serializer=key_serializer,
                         value_serializer=value_serializer)

# 查询需要输出到Kafka的数据
df = spark.sql("SELECT name, age FROM people")

# 将数据转换为JSON格式
df_json = df.select(to_json(struct(df.columns)).alias("value"))

# 遍历查询结果，将每条记录写入Kafka
for row in df_json.collect():
    record = row["value"]
    producer.send(topic=topic, value=record)

# 关闭KafkaProducer
producer.close()

# 关闭SparkSession
spark.stop()