使用spark结构化流媒体从kafka中读取数据，总是会出现超时问题

72qzrwbm 于 2021-06-07 发布在 Kafka

关注(0)|答案(1)|浏览(356)

这是我用spark结构化流媒体读取Kafka数据的代码，

//ss:SparkSession is defined before. 
import ss.implicits._
val df = ss
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", kafka_server)
  .option("subscribe", topic_input)
  .option("startingOffsets", "latest")
  .option("kafkaConsumer.pollTimeoutMs", "5000")
  .option("failOnDataLoss", "false")
  .load()

这是错误代码，

Caused by: java.util.concurrent.TimeoutException: Cannot fetch record xxxx for offset in 5000 milliseconds

如果我把5000扩大到10000，这个错误仍然会发生。我用谷歌搜索这个问题。关于这个问题似乎没有太多的相关信息。
以下是sbt文件中与此问题相关的部分。

libraryDependencies += "org.apache.spark" %% "spark-sql-kafka-0-10" % "2.3.0" exclude ("org.apache.kafka", "kafka-clients")
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.11.0.0"

apache-kafka apache-spark spark-structured-streaming

来源：https://stackoverflow.com/questions/49526632/using-spark-structured-streaming-to-read-data-from-kafka-issue-of-over-time-is

1条答案

按热度按时间

dzjeubhm1#

我也有这个错误。
我看了Kafka索的源代码，什么都没有。
我猜kafka连接器有问题，因此我在“spark-sql-kafka-0-10_.11”包中排除了kafka客户端，并添加了一个新的依赖项，如下所示：

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>2.3.0</version>
        <scope>compile</scope>
        <exclusions>
            <exclusion>
                <artifactId>kafka-clients</artifactId>
                <groupId>org.apache.kafka</groupId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.10.2.1</version>
    </dependency>

现在起作用了。希望有帮助。
我创建了一个jira问题来报告这个问题：https://issues.apache.org/jira/browse/spark-23829
2018年12月17日更新：spark 2.4和kafka2.0解决了问题。

赞(0）回复(0）举报 2021-06-07

我来回答

使用spark结构化流媒体从kafka中读取数据，总是会出现超时问题

1条答案

相关问题

热门标签

最新问答