如何将从 RabbitMQ 队列接收的 JSON 转换为 CSV？答案

【问题标题】：How to convert JSON received from a RabbitMQ Queue to CSV?如何将从 RabbitMQ 队列接收的 JSON 转换为 CSV？
【发布时间】：2020-03-02 12:11:10
【问题描述】：

目前我正在使用来自我组织内的 RabbitMQ 队列的消息。每天我都需要将收到的所有消息推送到一个 csv 中，该 csv 最终将作为数据仓库中的一个表登陆。

代码始终在监听队列，理想情况下我希望将数据流式传输到 csv。

#callback funtion on receiving messages
def onMessage(channel, method, properties, body):
    print(body)

while True:
    try:
        #connect
        credentials = pika.PlainCredentials(username, password)
        connection = pika.BlockingConnection(pika.ConnectionParameters(host = server, port = port, virtual_host = vhost, credentials = credentials))


channel = connection.channel()
channel.basic_consume(on_message_callback = onMessage, queue = queueName, auto_ack = True)
        channel.start_consuming()

我开始消费队列后收到的输出如下：这是接收到的一行数据。它基本上返回一个 json 对象，但是 b'{"metrics": 在使用 json 对象时需要删除。

b'{"metrics":[{"ci_id":"SPN-EQSHATA1","client_id":"39956e6fdb256757567567433333193a","name":"deviceHealthScore","source_id":"Global","source_management_platform": "XXX","timestamp":1582886099642,"unit":"count","value":"10.0"},{"ci_id":"SPN-EQSHATA1","client_id":"39956e6fdb256757567567433333193a","name": "configAssuranceScore","source_id":"Global","source_management_platform":"XXX","timestamp":1582886099325,"unit":"count","value":"1.0"},{"ci_id":"SPN -EQSHATA1","client_id":"39956e6fdb256757567567433333193a","name":"imageAssuranceScore","source_id":"Global","source_management_platform":"XXX","timestamp":1582886099325,"unit":"count", "value":"1.0"},{"ci_id":"SPN-EQSHATA1","client_id":"39956e6fdb256757567567433333193a","name":"vulnerabilityAssuranceScore","source_id":"Global","source_management_platform":"XXX ","timestamp":1582886099325,"unit":"count","value":"10.0"},{"ci_id":"SPN-EQSHATA1","client_id":"39956e6fdb256757567567433333193a","name":"overallAssuranceScore ","source_id":"全球","所以urce_management_platform":"XXX","timestamp":1582886099642,"unit":"count","value":"5.5"}],"emr_published_on":1582886099642}'

【问题讨论】：

您尝试解析 json 并写入 csv 文件有什么具体问题？ csv文件的结构是什么？

标签： python json pandas rabbitmq pika

【解决方案1】：

b'...' 只是意味着你得到了一个 json 模块可以愉快地处理的字节字符串。你会得到一个字典，对于metrics 键，它具有字典列表的价值。该列表可以直接提供一个 DataFrame。

这意味着您可以像这样简单地处理它：

df = pd.DataFrame(json.loads(body)['metrics'])

【讨论】：

非常感谢 Serge，像魅力一样工作。我将输出集中到一个 CSV 中，以便最终附加。