confluent kafka

 

 

https://www.confluent.io/blog/introduction-to-apache-kafka-for-python-programmers/?utm_medium=sem&utm_source=google&utm_campaign=ch.sem_br.brand_tp.prs_tgt.confluent-brand_mt.mbm_rgn.apac_lng.eng_dv.all_con.confluent-python&utm_term=%2Bconfluent%20%2Bpython&creative=&device=c&placement=&gclid=Cj0KCQjw1ouKBhC5ARIsAHXNMI9W1SnTGyAdPKPPQof4PprZhrs-n7woxaHJEPQMhthz1bW14rwsxlYaAh9wEALw_wcB 

Introduction to Apache Kafka® for Python Programmers | Confluent

 

 

 

https://data-newbie.tistory.com/252

[Python] confluent-Kafka 연습하기

 

 

 

from confluent_kafka import Consumer, KafkaError

settings = {
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'mygroup',
    'client.id': 'myclient',
    'enable.auto.commit': False,
    'session.timeout.ms': 6000,
    'default.topic.config': {'auto.offset.reset': 'earliest'}
}

c = Consumer(settings)

c.subscribe(['topic'])

try:
    while True:
        msg = c.poll(1)
        if msg is None:
            continue
        elif not msg.error():
            print('Received message: {0}:::{1}'.format(
                msg.offset(), msg.value()))
        elif msg.error().code() == KafkaError._PARTITION_EOF:
            print('End of partition reached {0}/{1}'
                  .format(msg.topic(), msg.partition()))
        else:
            print('Error occured: {0}'.format(msg.error().str()))

except KeyboardInterrupt:
    pass

finally:
    c.close()