PySpark写入Confluent Kafka报错及方案正确性、配置需求咨询

1. AttributeError报错根因 #

foreachPartition的回调函数接收的参数是当前分区的行迭代器（属于itertools.chain类型），不是DataFrame对象，自然没有selectExpr方法。
同时这属于错误用法：Spark原生已经内置了Kafka写入数据源，不需要手动遍历分区写入，原生实现已经做了连接复用、异常重试等容错逻辑，比自定义实现更稳定高效。

2. Kafka超时报错根因 #

你的配置完全没有添加SASL PLAIN认证的相关参数，Confluent Cloud的Kafka服务要求必须携带有效认证信息才能访问，无认证的请求会被直接拒绝连接，无法拉取主题元数据，超过60秒就会抛出超时异常。

直接调用Spark原生的Kafka写入接口，补充完整的认证配置即可：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.appName("ConfluentKafkaWrite").getOrCreate()

# 直接对源DataFrame做转换后写入，不需要调用foreachPartition
dataframe.selectExpr("to_json(struct(*)) AS value") \
    .write \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "xxx.aws.confluent.cloud:9092") \
    .option("topic", "test_topic") \
    # SASL PLAIN认证必填配置
    .option("kafka.security.protocol", "SASL_SSL") \
    .option("kafka.sasl.mechanism", "PLAIN") \
    .option("kafka.sasl.jaas.config", 
            'org.apache.kafka.common.security.plain.PlainLoginModule required username="你的Confluent API Key" password="你的Confluent API Secret";') \
    # 可选调优参数，降低超时概率
    .option("kafka.request.timeout.ms", "120000") \
    .option("kafka.metadata.max.age.ms", "300000") \
    .save()

• 替换配置中的你的Confluent API Key和你的Confluent API Secret为你在Confluent Cloud控制台生成的实际密钥对，注意sasl.jaas.config配置的末尾必须带分号，引号规则不要写错。
• 需要提前确认Spark集群的安全组、网络策略允许节点访问Confluent Cloud的9092端口，网络连通性是正常写入的基础。
• 如果是结构化流场景写入Kafka，只需要把.write替换为.writeStream，加上对应的触发配置即可，核心的Kafka参数配置通用。

内容的提问来源于stack exchange，提问作者Ajay Kumar