问题描述
{
"ID": 300,"Name": " TEST","Value": [
{
"Details": [
{
"Name": "TEST1","Value": "XXXXXX"
},{
"Name": "TEST2","Value": "dddddddd"
}
],"Time": [ 1600358400,1600358700,1600359000],"Values": [ 0,0]
}
]
}
我想展平json以便能够获得类似列表
我使用了itertools groupby，但是无法达到预期的结果。它正在水平展平。
到目前为止我尝试过的代码
from itertools import groupby
import json
def myflatten(d,depth=0):
rv = [({},depth)]
if isinstance(d,dict):
for k,v in d.items():
if not isinstance(v,dict) and not isinstance(v,list):
for i in rv:
i[0][k] = v
else:
for (vv,_depth) in myflatten(v,depth+1):
rv.append((rv[-1][0].copy(),_depth))
for kkk,vvv in vv.items():
rv[-1][0][kkk] = vvv
elif isinstance(d,list):
for v in d:
rv.append((rv[-1][0].copy(),depth+1))
for (vv,_) in myflatten(v,depth+1):
for kkk,vvv in vv.items():
rv[-1][0][kkk] = vvv
for i,_depth in rv:
yield i,_depth
out = []
a = {
"ID": 300,0]
}
]
}
for v,g in groupby(sorted(myflatten(a),key=lambda k: -k[1]),lambda k: k[1]):
out.extend(i[0] for i in g)
break
print(out)
有人可以帮助垂直而不是水平拉平嵌套的json / dict / list吗？最终目标是能够将数据存储在RDBMS中，而不必无限增加列数，而不必增加行数。
解决方法
由于在您的情况下，您需要将行（或记录）存储在数据库中，因此可以使用一个函数来生成字典列表，然后将其迭代添加到数据库中。另外，由于您的代码似乎过于嵌套，因此您可以考虑以下代码片段：
def flatten(dictionary):
my_list = []
_id = dictionary["id"]
name = dictionary["Name"]
for obj in dictionary["Value"]:
details = obj["Details"]
time = obj["Time"]
vals = obj["Values"]
for i in range(len(time)):
for (index,detail_obj) in enumerate(details):
my_list.append({
"ID": _id,"Name": name,"Details.ID": index,"Details.Name": detail_obj["Name"]
"Details.Value": detail_obj["Value"],"Time.ID": i,"Time.Time": time[i],"Time.Value": vals[i]
})
return my_list
注意：：此功能将一次处理一个嵌套词典（例如您在问题中提供的词典）。因此，如果您有多个这样的嵌套字典，则可能需要为每个嵌套字典调用此函数。
,为了概括这一点，我们需要将所有内容展平，但是将迭代器返回到列表中。然后，我们将这些迭代器压缩在一起，以便它们都以相同的速度前进。但是这些迭代器可能是必须将其平整的指令，记住它们来自哪个层次。
此外，请注意，外部字典的ID，名称，值不受此限制，并且不包含在递归命名方案中，因此我们将编写一个顶级处理程序来解析它们，然后设置其余处理程序。而且“ Details.ID”不是数据的一部分，因此我对其进行了硬编码，其ID取自上述列表中的枚举。
我承认，对扁平化函数进行泛型处理来处理字典，列表和标量数据的任意嵌套对我来说有点麻烦，因为我们需要为多个不同类型的链结迭代器并选择一致的返回类型。输入。太多...
相反，我在有限的程度上使用了数据的结构，并假设只有一个级别包含列表。我只在数据内展平。这大大简化了问题，同时仍然可以在所有级别接受任意名称和值：
import itertools,json,typing,pprint
# top level function starts recursion
def parse(data):
fixed_fields = {k:v for k,v in data.items() if not isinstance(v,typing.Iterable)}
for testcase in data['Value']:
for testcase in data['Value']:
for record in parse_testcase(testcase):
record.update(fixed_fields)
yield record
def parse_testcase(testcase):
names = []
values = []
for key,value in testcase.items():
names.append(key)
values.append(itertools.chain(value))
for details_id,row in enumerate(zip(*values)):
record = {'Details.ID': details_id}
for name,value in zip(names,row):
if isinstance(value,dict):
flatten(name,value,record)
else:
record[name] = value
yield record
def flatten(parent_key,details,result):
for key,value in details.items():
keyname = get_keyname(parent_key,key)
if isinstance(value,dict):
flatten(keyname,result)
else:
result[keyname] = value
def get_keyname(parent_key,key):
if parent_key:
return '.'.join((parent_key,key))
return key
text = """{
"ID": 300,"Name": " TEST","Value": [
{
"Details": [
{
"Name": "TEST1","Value": "XXXXXX"
},{
"Name": "TEST2","Value": "DDDDDDDD"
}
],"Time": [ 1600358400,1600358700,1600359000],"Values": [ 0,0]
}
]
}"""
for record in parse(json.loads(text)):
pprint.pprint(record)
由于您有2条“详细信息”记录和3条其他记录，因此我不得不怀疑您的预期输出是错误的。我唯一可以理解的解析规则是同步推进所有列表，因此减少了多余的时间和值的测量。如果您可以帮助我理解解析规则，以便您了解如何获得四行，我们可以重新进行讨论。我得到了：
{'Details.ID'：0，
'Details.Name'：'TEST1'，
'Details.Value'：'XXXXXX'，
'ID'：300，
“时间”：1600358400，
'值'：0}
{'Details.ID'：1，
'Details.Name'：'TEST2'，
'Details.Value'：'DDDDDDDD'，
'ID'：300，
'时间'：1600358700，
'值'：0}
我还注意到，您的输出具有由输入确定的任意方案。也许输出一个csv文件并使其适合关系数据库是一项单独的任务。