我使用spark-sql-2.4.x版本的kafka客户端。
即使在设置了使用者配置参数max.partition.fetch.bytes和max.poll.records之后
设置不正确,显示默认值如下
Dataset<Row> df = sparkSession
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", server1)
.option("subscribe", TOPIC1)
.option("includeTimestamp", true)
.option("startingOffsets", "latest")
.option("max.partition.fetch.bytes", "2097152") // default 1000,000
.option("max.poll.records", 6000) // default 500
.option("metadata.max.age.ms", 450000) // default 300000
.option("failOnDataLoss", false)
.load();
在启动使用者时,它仍显示在日志中,如下所示:
[Executor task launch worker for task 21] INFO org.apache.kafka.clients.consumer.ConsumerConfig - ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = none
check.crcs = true
client.id =
connections.max.idle.ms = 540000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
heartbeat.interval.ms = 3000
interceptor.classes = null
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
正确的设置方法是什么?
1条答案
按热度按时间7rfyedvj1#
根据文件:
Kafka自己的配置可以通过带有Kafka的datastreamreader.option来设置。前缀,例如stream.option(“kafka.bootstrap.servers”,“host:port"). 有关可能的kafka参数,请参阅kafka consumer config docs以获取与读取数据相关的参数,以及kafka producer config docs以获取与写入数据相关的参数。
我认为你需要在你的选择中加上“Kafka”,比如: