Spark Structured Streaming无法修改max.poll.records（显示为1）的问题咨询

1. 如何将max.poll.records修改为期望值？ #

你无法通过kafka.max.poll.records参数直接修改底层KafkaConsumer的该配置，因为Spark Structured Streaming的Kafka源会内部强制覆盖这个参数。要控制每次拉取的记录数量，你需要使用Spark提供的专属参数：maxOffsetsPerTrigger。

修改你的代码，添加这个参数到readStream的配置中：

df_stream = spark.readStream.format('kafka') \
        .option("kafka.security.protocol", "SSL") \
        .option("kafka.ssl.truststore.location", ssl_truststore_location) \
        .option("kafka.ssl.truststore.password", ssl_truststore_password) \
        .option("kafka.ssl.keystore.location", ssl_keystore_location) \
        .option("kafka.ssl.keystore.password", ssl_keystore_password) \
        .option("kafka.bootstrap.servers", kafkaBrokers) \
        .option("subscribe", topic) \
        .option("startingOffsets", "latest") \
        .option("failOnDataLoss", "false") \
        .option("kafka.metadata.max.age.ms", "1000") \
        .option("kafka.ssl.keystore.type", "PKCS12") \
        .option("kafka.ssl.truststore.type", "PKCS12") \
        .option("kafka.max.poll.interval.ms", 600000) \
        .option("maxOffsetsPerTrigger", 7000)  # 用这个参数控制每次触发拉取的最大记录数
        .load()

这个参数定义了每个微批处理触发时，从Kafka拉取的最大偏移量总数，这才是Spark层面控制拉取批次大小的正确方式。

2. 为何该值显示为1而非默认的500？ #

Spark Structured Streaming的Kafka源在初始化底层KafkaConsumer时，会强制将max.poll.records设置为1，这是Spark的设计决定：

• 为了实现精确的偏移量追踪和故障恢复，Spark需要逐个确认每个记录的偏移量，避免因Kafka一次性拉取大量记录导致偏移量范围无法精确控制。
• Spark自己管理批次的大小（通过maxOffsetsPerTrigger），不需要依赖Kafka的max.poll.records参数来控制批量，因此会覆盖该参数的值。

所以你看到的max.poll.records=1是Spark内部设置的，和Kafka原生默认值无关，用户通过kafka.max.poll.records设置的值不会生效。

另外，针对你遇到的poll timeout警告，除了调整maxOffsetsPerTrigger减少每次处理的记录数，还可以：

• 优化数据处理逻辑，缩短每个微批的处理时间
• 确认max.poll.interval.ms的设置足够覆盖你的处理时长（当前600000的配置是合理的）

内容的提问来源于stack exchange，提问作者Karan Alang