如何对Kafka轮询消费的数据按指定时间戳条件进行过滤

Kafka消息本身自带服务端生成的时间戳，你可以根据需求选择两种实现方式：

方案1：消费前定位到指定时间戳对应offset（推荐，性能更高） #

这个方式会直接跳过早于目标时间的消息，不需要拉取无效数据：

import json
import time
from kafka import KafkaConsumer, TopicPartition
from pandas import json_normalize

bootstrap_servers = ['server.com']
topicName = 'topicname'
consumer = KafkaConsumer(
    group_id='topic',
    bootstrap_servers=bootstrap_servers,
    auto_offset_reset='earliest',
    consumer_timeout_ms=20000
)

# ------------------新增时间戳定位逻辑------------------
# 替换成你需要的起始时间，示例是2024-01-01 00:00:00的毫秒级时间戳
target_start_timestamp = int(time.mktime(time.strptime("2024-01-01 00:00:00", "%Y-%m-%d %H:%M:%S")) * 1000)

# 获取该topic的所有分区
partitions = [TopicPartition(topicName, p) for p in consumer.partitions_for_topic(topicName)]
# 给每个分区传入目标时间戳，查询对应offset
timestamp_to_search = {tp: target_start_timestamp for tp in partitions}
offset_rsp = consumer.offsets_for_times(timestamp_to_search)

# 把每个分区的消费位置定位到查询到的offset
for tp, offset_info in offset_rsp.items():
    if offset_info is not None:
        consumer.seek(tp, offset_info.offset)
    else:
        # 如果该分区没有晚于目标时间的消息，直接跳到分区末尾
        consumer.seek_to_end(tp)
# -----------------------------------------------------

data_list = []
for message in consumer:
    # 如果需要加结束时间限制，也可以在这里判断message.timestamp < 结束时间戳就break循环
    print(message)
    data = json.loads(message.value)
    df = json_normalize(data)
    data_list.append(df)

注意：如果你需要按消息体里自带的业务时间字段筛选，而不是Kafka服务端生成的消息时间，用下面的方案2

方案2：消费后本地过滤（灵活度更高） #

适合需要多条件过滤、或者用业务时间戳筛选的场景，直接在消费循环里加判断即可：

import json
from kafka import KafkaConsumer
from pandas import json_normalize

bootstrap_servers = ['server.com']
topicName = 'topicname'
consumer = KafkaConsumer(
    topicName,
    group_id='topic',
    bootstrap_servers=bootstrap_servers,
    auto_offset_reset='earliest',
    consumer_timeout_ms=20000
)

# 定义你的时间范围，示例是业务时间在2024-01-01到2024-01-02之间的消息
start_time = "2024-01-01 00:00:00"
end_time = "2024-01-02 00:00:00"

data_list = []
for message in consumer:
    print(message)
    data = json.loads(message.value)
    # 假设你的消息体里业务时间字段是event_time，格式是%Y-%m-%d %H:%M:%S
    event_time = data.get("event_time")
    if not event_time or event_time < start_time or event_time >= end_time:
        # 不符合时间条件直接跳过
        continue
    df = json_normalize(data)
    data_list.append(df)

内容的提问来源于stack exchange，提问作者Kevin Nash