我正在尝试通过火花流从Kafka经纪人那里阅读,但是我遇到了一些问题。
def spark_streaming_from_STABLE_kafka_topic():
conf = SparkConf().setMaster("spark://antonis-dell:7077").setAppName("Kafka_Spark")
sc = SparkContext(conf=conf)
sc.setLogLevel("WARN")
ssc = StreamingContext(sc, 2)
topic = "stable_topic"
kvs = KafkaUtils.createDirectStream(ssc,
[topic],
{"metadata.broker.list": "my-broker",
"auto.offset.reset": "smallest"},
keyDecoder=lambda x: x,
valueDecoder=lambda x: x
)
lines = kvs.window(2, 2).map(lambda x: x[1])
lines.pprint()
return ssc
if __name__ == "__main__":
ssc = StreamingContext.getOrCreate('/home/antonis/Desktop/tmp/checkpoint_v06', lambda: spark_streaming_from_STABLE_kafka_topic())
ssc.start()
ssc.awaitTermination()
上面的代码除了空批处理以外,不提取任何东西:
-------------------------------------------
Time: 2020-05-29 09:32:38
-------------------------------------------
-------------------------------------------
Time: 2020-05-29 09:32:40
-------------------------------------------
主题stable_topic
包含固定大小的数据。它不会改变。我还有另一个每秒接收数据的主题。如果我使用本主题而不是stable_topic
并删除了"auto.offset.reset": "smallest"
,则代码将获取数据。
我认为{"auto.offset.reset": "smallest"}
出了点问题,但我无法弄清楚。
现在有人在做什么我做错了吗?
[在更高版本中,smallest
被earliest
代替。确保检查使用的版本的文档。
此外,如果使用者组已经使用了主题auto.offset.reset
中的某些数据,则stable_topic
配置将不会生效。因此,您可以考虑在流作业中更改group.id
。