代码之家 › 专栏 › 技术社区 › dportman

卡夫卡到熊猫的无火花数据帧

apache-kafka pandas json python

dportman · 技术社区 · 6 年前

我正在阅读卡夫卡主题的流式数据,我想将其中的一些部分存储在一个数据框中。

from confluent_kafka import Consumer, KafkaError

c = Consumer({
    'bootstrap.servers': "###",
    'group.id': '###',
    'default.topic.config': {
'auto.offset.reset': 'latest' }
})

c.subscribe(['scorestore'])

while True:
    msg = c.poll(1.0)

    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            print(msg.error())
            break

    print('Received message: {}'.format(msg.value().decode('utf-8')))

c.close()

收到的消息是json

{
  "messageHeader" : {
    "messageId" : "4b604b33-7256-47b6-89d6-eb1d92a282e6",
    "timestamp" : 152520000,
    "sourceHost" : "test",
    "sourceLocation" : "test",
    "tags" : [ ],
    "version" : "1.0"
  },
  "id_value" : {
    "id" : "1234",
    "value" : "333.0"
  }
}

例如,我正在尝试创建一个包含timestamp、id和value列的数据帧

    timestamp   id  value
0   152520000   1234    333.0

有没有一种方法可以在不解析json消息并将所需的值逐行附加到数据帧的情况下实现这一点?

1 回复 | 直到 6 年前

migjimen 6 年前

我提出的解决方案可能有点棘手。假设您的JSON消息位于一个名为“msg\u str”的字符串中:

import pandas as pd

msg_str = '{  "messageHeader" : { "messageId" : "4b604b33-7256-47b6-89d6-eb1d92a282e6",    "timestamp" : 152520000,    "sourceHost" : "test",    "sourceLocation" : "test",    "tags" : [ ],    "version" : "1.0"  },  "id_value" : {    "id" : "1234",    "value" : "333.0"  }}'


#first create a dataframe with read_json
p = pd.read_json(msg_str)
# Now you have a dataframe with two columns. Where a column has a value, the other 
# has a NaN. Now create a new column only with the values which are not 'NaN'
p['fussion'] = p['id_value'].fillna(p['messageHeader'])
# Delete columns 'id_value' and 'messageHeader' as you don't need them anymore
p = p[['fussion']].reset_index()
# Create a temporal column only to be the index to do a pivot
p['tmp'] = 0
# Do the pivot to convert rows into columns
p = p.pivot(index = 'tmp' ,values='fussion', columns='index')
# Finally get the columns that you are interested in
p = p.reset_index()[['timestamp','id','value']]

print(p)

结果:

index  timestamp    id value
0      152520000  1234   333

然后,您可以将此数据帧附加到正在累积结果的数据帧中。

也许有一个最简单的解决方案,但如果不是这样,我希望它能帮助你。

推荐文章

Mainland · Python数据帧规范化值错误:列的长度必须与键相同

1 年前

user026 · 如何根据特定窗口的平均值(行数)创建新列?

1 年前

rpn · 如何在列[1]中连续第二次出现“0”时返回列[0]的值

1 年前

asmgx · 为什么合并数据帧不能按照python中的预期方式工作

1 年前

Gtoth · 如何分割Pandas DataFrame中包含多个日期的两个时间戳之间的差异

1 年前

Domarius · 使用loc为多行设置多列值

1 年前

Swastik Bhattacharyya · 如何在同一类别类型的多列上运行get_dummies()函数?

1 年前

DrZoidberg09 · 如何在字典列表中创建一个新关键字,该关键字是另一个关键字的总和?

1 年前

armstrong3701 · 如何有效地处理熊猫数据框中缺失的数据并计算条件统计?

1 年前

msts1906 · 大熊猫向乳胶的适当多品种出口

1 年前