我希望使用sparkstreaming保持作业运行

Question

有没有可能让流作业一直运行？大约24小时后，它发出此错误并停止处理。我不太确定该怎么处理这件事。

21/01/01 00:03:30 WARN KafkaOffsetReader [stream execution thread for [id =17bf-45aa-a9cd-2f77ec14df61, runId = 43c1-a932-d9f790996a6e]]: Retrying to fetch latest offsets because of incorrect offsets
21/01/01 07:17:04 ERROR RawSocketSender [MdsLoggerSenderThread]: org.fluentd.logger.sender.RawSocketSender
java.net.SocketException: Broken pipe (Write failed)

ssc.awaitTermination()

上面的代码不是一直都在运行吗？

Answer 1

原因:您的kafka队列中没有消息可供消费。

增加awaitTermination()的重试和最大重试时间。

ie等待3000000毫秒=等待消息5分钟

  ssc.awaitTermination(100000)

注意:根据您的环境更改该值。这是最大持续时间，将至少有一条新消息到达kafka队列。