python kafka按时间戳消费

淡淡的烟草味﹌ 2022-12-22 11:17 195阅读 0赞

> 公司业务需求,获取存入kafka大于某个时间段的数据

# -*- coding: utf-8 -*-
    # @Time : 2020/11/20 09:01
    # @Author :
    
    from confluent_kafka import Consumer, TopicPartition
    
    KAFKASERVERS = 'xx.xx.xx.6:9092,xx.xx.xx.4:9092,xx.xx.xx.7:9092'
    GROUPNAME = 'my_group_name'
    
    c = Consumer({ 
        'bootstrap.servers': KAFKASERVERS,
        'group.id': GROUPNAME,
        'auto.offset.reset': 'earliest'
    })
    
    # 主题名
    topic = 'my_topic_name'
    
    # 获取当前topic存在多少个分区
    cluster_data = c.list_topics(topic=topic)
    topic_data = cluster_data.topics[topic]
    available_partitions = topic_data.partitions
    
    # c.subscribe([topic])
    # 把每个partition的offset设置到指定时间戳下,即获取大于改timestamp入库kafka的数据
    # 注意这里的时间戳位数
    timestamp = 1605756666790
    tps = [TopicPartition(topic, tp, timestamp) for tp in range(len(available_partitions))]
    offsets = c.offsets_for_times(tps)
    c.assign(offsets)
    
    while True:
        # 阻塞等待消息的最大时间
        msg = c.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print("Consumer error: {}".format(msg.error()))
            continue
        # 获取该数据入kafka时间戳
        kafka_timestamp = msg.timestamp()[1]
        # print(kafka_timestamp)
        # 消费kafka相应数据
        print('Received message: {}'.format(msg.value().decode('utf-8')))
    c.close()

[Kafka-python获取主题的分区数][Kafka-python]

[Kafka-python]: https://xbuba.com/questions/29612228

发表评论取消回复

表情：

评论列表（有 0 条评论，195人围观）

还没有评论，来说两句吧...

相关阅读

相关 python 时间转时间戳，时间戳转时间

import datetime import time class Util: @staticmethod

快来打我*/ 2024年03月30日 17:01/ 0 赞/ 92 阅读

相关 Kafka：指定时间消费

需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据，怎么处理？ import org.apache.kaf

朴灿烈づ我的快乐病毒、/ 2024年03月25日 22:51/ 0 赞/ 72 阅读

相关 python时间戳转换成时间

python问题，字符串的时间可以直接转换为时间戳吗？ ts='2020-04-0113:43:36' dt=datetime.datetime.strptime(ts

系统管理员/ 2023年09月25日 11:18/ 0 赞/ 89 阅读

相关 flink 根据时间消费kafka

kafka版本：0.11 flink版本：1.9 FlinkKafkaConsumer011中提供了很多方法确定如何消费kafka消息： setStart

冷不防/ 2023年05月31日 10:25/ 0 赞/ 31 阅读

相关 python kafka按时间戳消费

> 公司业务需求,获取存入kafka大于某个时间段的数据 -- coding: utf-8 -- @Time : 2020/11/20 09:01

淡淡的烟草味﹌/ 2022年12月22日 11:17/ 0 赞/ 196 阅读

相关 python time时间,日期,时间戳处理

1.将字符串的时间转换为时间戳方法: a = "2013-10-10 23:40:00" 将其转换为时间数组

蔚落/ 2022年09月29日 05:25/ 0 赞/ 322 阅读

相关五.Kafka消息的时间戳

kafka在消息中增加了一个时间戳字段和时间戳类型。目前支持的时间戳类型有两种： CreateTime 和 LogAppendTime 前者表示producer创建这条消息的时

痛定思痛。/ 2022年07月13日 18:30/ 0 赞/ 221 阅读

相关 python kafka生产与消费

消费者： from kafka import KafkaConsumer consumer=KafkaConsumer(bootstrap\_serve

红太狼/ 2022年05月26日 12:21/ 0 赞/ 230 阅读

相关 iOS 中表格按时间戳分组排序

最近在写公司的项目的时候,遇到一个需求,后台返回一些交易列表,前端 app 需要根据后台返回的数据 Model 里面的时间对数据分组排序,大概是这样 > 具体需求是本年只显示

女爷i/ 2022年01月12日 01:39/ 0 赞/ 411 阅读

相关什么是时间戳？用python如何显示时间戳

什么是时间戳？用python如何显示时间戳时间戳（Unix timestamp）：简单来说：从格林威治时间（1970年01月01日08时00分00秒）开始到当前时间的...

- 日理万妓/ 2021年05月20日 14:19/ 0 赞/ 889 阅读