python kafka 工具类

以下是Python Kafka工具类的介绍和演示：

Python Kafka工具类是一个用于操作Kafka的Python库，它提供了一些常用的Kafka操作方法，例如获取Kafka TOPIC中的偏移量等。下面是两个常用的方法：

1.获取Kafka TOPIC中在每个时间点的偏移量

from kafka import KafkaConsumer, TopicPartition

def kafka_get_offsets_range(consumer, topic_name):
    partitions = consumer.partitions_for_topic(topic_name)
    partitions = [TopicPartition(topic_name, p) for p in partitions]
    begin_offsets = consumer.beginning_offsets(partitions)
    end_offsets = consumer.end_offsets(partitions)
    return [(p.partition, begin_offsets[p], end_offsets[p]) for p in partitions]

2.获取Kafka TOPIC在指定时间段内的偏移量

from datetime import datetime, timedelta
from kafka import KafkaConsumer, TopicPartition

def kafka_get_offset_by_period(consumer, topic_name, start_time, end_time, distance):
    partitions = consumer.partitions_for_topic(topic_name)
    partitions = [TopicPartition(topic_name, p) for p in partitions]
    result = {}
    for p in partitions:
        result[p.partition] = []
        offset = consumer.offsets_for_times({p: int(start_time.timestamp() * 1000)})[p].offset
        end_offset = consumer.offsets_for_times({p: int(end_time.timestamp() * 1000)})[p].offset
        while offset < end_offset:
            result[p.partition].append(offset)
            offset += 1
            if offset % distance == 0:
                consumer.seek(p, offset)
    return result