如何使用Python在kafka使用者中聚合json数据？

Question

我的数据在Kafka事务处理主题中是这样的：

ConsumerRecord(Topic='transactions',partition=0,offset=4,timestamp=1591277946736,timestamp_type=0,key=none,value={'transaction_id':'4952940859','account_number':14,'transaction_reference':'44291','transaction_datetime':'2020-06-04T19:09:06.736128','amount':2.82},headers=[],checksum=none,

ConsumerRecord(Topic='transactions',partition=0,offset=5,timestamp=1591277946737,timestamp_type=0,key=none,value={'transaction_id':'0193362270','account_number':12,'transaction_reference':'96312','transaction_datetime':'2020-06-04T19:09:06.736128','amount':766.95},headers=[],checksum=none,

到目前为止编写的消费者守则是：

consumer = KafkaConsumer(bootstrap_servers='localhost:9092',
                                 auto_offset_reset='earliest',
                                 value_deserializer=lambda m: json.loads(m.decode('utf-8')))
consumer.subscribe(['Transactions'])
for message in consumer:
            print (message)

我希望输出像元组(account_number，sum(amount))，我如何实现这一点？

Answer 1

我认为字典可能比元组更有用。DefaultDict将非常适合此过程

from collections import defaultdict

accounts = defaultdict(int)

for message in consumer:
    payload = message.value
    account = payload['account_number']
    amount = payload['amount']

    accounts[account] += amount


print(accounts)

defaultdict(<class 'int'>,{
  "14": 263.75,
  "12": 766.95
})

要获得可能要查找的元组，可以在循环之后遍历accounts.items()：

for info in accounts.items():
    print(info)

("14", 263.75)
("12", 766.95)