来自kafka的json无法转换为pandas答案

【问题标题】：json from kafka cant convert to pandas来自kafka的json无法转换为pandas
【发布时间】：2021-04-28 09:36:16
【问题描述】：

嘿，我有这样的代码来消耗 kafka 数据

 bootstrap_servers = ['localhost:9092']
 topicName = 'testapp5'
 consumer = KafkaConsumer (topicName, group_id ='group1',bootstrap_servers = bootstrap_servers)
 for msg in consumer:
       print("Topic Name=%s,Message=%s"%(msg.topic,msg.value))

然后我想加载数据

  message = json.loads(msg.value)

输出：

 {'request_id': 'f84c55fd-c730-49ba-83b2-47b04643b706',
'data': {'age': 24,
'workclass': 'Self-emp-not-inc',
'fnlwgt': 188274,
'education': 'Bachelors',
'marital_status': 'Never-married',
'occupation': 'Sales',
'relationship': 'Not-in-family',
'race': 'White',
'gender': 'Male',
'capital_gain': 0,
'capital_loss': 0,
'hours_per_week': 50,
'native_country': 'United-States',
'income_bracket': '<=50K.'}}

然后我想将数据更改为熊猫数据框

  row = pd.DataFrame(message, index=[0])

和输出：

我应该怎么做才能使来自 kafka 的 json 可以通过 pandas 数据框访问？谢谢之前

【问题讨论】：

使用pd.json_normalize(message)
你的数据框只有一行，为什么要使用 pandas？
应该使用哪些建议？ @OneCricketeer
您只需要json.loads 获取一条消息，正如您在问题中所述，您可以将其作为value_deserializer 传递给消费者。当你没有一批消息时，我不清楚为什么你会想要熊猫。如果您确实想要一个记录批次和数据帧，那么带有结构化流的 pyspark 可以实现这一点，否则您需要定期将记录累积到一个列表中，然后将该列表刷新到 pandas 数据帧中

标签： python json python-3.x pandas apache-kafka

【解决方案1】：

这个最简单的方法是使用json_normalize。如果您只想要 data，您可以使用 dict 键使用 pd.DataFrame。

js = {'request_id': 'f84c55fd-c730-49ba-83b2-47b04643b706',
'data': {'age': 24,
'workclass': 'Self-emp-not-inc',
'fnlwgt': 188274,
'education': 'Bachelors',
'marital_status': 'Never-married',
'occupation': 'Sales',
'relationship': 'Not-in-family',
'race': 'White',
'gender': 'Male',
'capital_gain': 0,
'capital_loss': 0,
'hours_per_week': 50,
'native_country': 'United-States',
'income_bracket': '<=50K.'}} 
# simplest....
pd.json_normalize(js)
# if requestid is not needed
pd.DataFrame(js["data"], index=[0])

【讨论】：