pySpark Kafka Direct Streaming更新Zookeeper / Kafka Offset

作者: 不见你
发布时间: 2024-12-17 06:03:36 (2月前)
转自：

3 条回复

0#
回复此人
布偶的表弟派大星丶 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 我遇到过类似的问题。你是对的，通过使用directStream，意味着直接使用kafka低级API，它没有更新读者偏移量。 scala / java有几个例子，但不适用于python。但是你自己很容易做到，你需要做的是： </p> <UL> <LI> 从开头的偏移读取 </LI> <LI> 最后保存偏移量 </LI> </UL> <P> 例如，我通过执行以下操作来保存redis中每个分区的偏移量： </p> <pre> <code> stream.foreachRDD(lambda rdd: save_offset(rdd)) def save_offset(rdd): ranges = rdd.offsetRanges() for rng in ranges: rng.untilOffset # save offset somewhere </code> </pre> <P> 然后在开始时，您可以使用： </p> <pre> <code> fromoffset = {} topic_partition = TopicAndPartition(topic, partition) fromoffset[topic_partition]= int(value) #the value of int read from where you store previously. </code> </pre> <P> 对于一些使用zk跟踪偏移量的工具，最好在zookeeper中保存偏移量。这一页： <a href="https://community.hortonworks.com/articles/81357/manually-resetting-offset-for-a-kafka-topic.html" rel="nofollow noreferrer"> https://community.hortonworks.com/articles/81357/manually-resetting-offset-for-a-kafka-topic.html </A> 描述如何设置偏移量，基本上，zk节点是： / consumers / [consumer_name] / offsets / [topic name] / [partition id] 因为我们正在使用directStream，所以你必须组成一个消费者名称。 </p> </DIV>

编辑
1#
回复此人
甲基蓝 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 我编写了一些函数来保存并使用python读取Kafka偏移量 <a href="https://github.com/python-zk/kazoo" rel="nofollow noreferrer"> 卡祖笛 </A> 图书馆。 </p> <P> 第一个获取Kazoo客户端单例的函数： </p> <pre> <code> ZOOKEEPER_SERVERS = "127.0.0.1:2181" def get_zookeeper_instance(): from kazoo.client import KazooClient if 'KazooSingletonInstance' not in globals(): globals()['KazooSingletonInstance'] = KazooClient(ZOOKEEPER_SERVERS) globals()['KazooSingletonInstance'].start() return globals()['KazooSingletonInstance'] </code> </pre> <P> 然后函数读取和写入偏移量： </p> <pre> <code> def read_offsets(zk, topics): from pyspark.streaming.kafka import TopicAndPartition from_offsets = {} for topic in topics: for partition in zk.get_children(f'/consumers/{topic}'): topic_partion = TopicAndPartition(topic, int(partition)) offset = int(zk.get(f'/consumers/{topic}/{partition}')[0]) from_offsets[topic_partion] = offset return from_offsets def save_offsets(rdd): zk = get_zookeeper_instance() for offset in rdd.offsetRanges(): path = f"/consumers/{offset.topic}/{offset.partition}" zk.ensure_path(path) zk.set(path, str(offset.untilOffset).encode()) </code> </pre> <P> 然后在开始流式传输之前，您可以从zookeeper读取偏移并将它们传递给 <a href="http://spark.apache.org/docs/2.1.0/api/python/pyspark.streaming.html#pyspark.streaming.kafka.KafkaUtils.createDirectStream" rel="nofollow noreferrer"> createDirectStream </A> 对于 <code> fromOffsets </code> 参数： </p> <pre> <code> from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import KafkaUtils def main(brokers="127.0.0.1:9092", topics=['test1', 'test2']): sc = SparkContext(appName="PythonStreamingSaveOffsets") ssc = StreamingContext(sc, 2) zk = get_zookeeper_instance() from_offsets = read_offsets(zk, topics) directKafkaStream = KafkaUtils.createDirectStream( ssc, topics, {"metadata.broker.list": brokers}, fromOffsets=from_offsets) directKafkaStream.foreachRDD(save_offsets) if __name__ == "__main__": main() </code> </pre> </DIV>

编辑

登录后才能参与评论