使用ApacheNIFI或Python脚本迭代Json数组

Question

我有一个Json和一个数组字段，我想迭代数组，并为数组中包含的每个属性将其拆分为新行或对象

我目前正在使用ApacheNIFI，但我也可以使用python脚本

我的输入数据是：

{
  "workorder_id" : "99999",
  "properties" : [ {
    "id" : "11",
    "propertyType" : {
      "id" : "55834595398",
      "name" : "action"
    },
    "stringValue" : "string01",
    "nodeValue" : null
  }, {
    "id" : "22",
    "propertyType" : {
      "id" : "55834595419",
      "name" : "Tipo"
    },
    "stringValue" : "string02",
    "nodeValue" : null
  }, {
    "id" : "33",
    "propertyType" : {
      "id" : "44",
      "name" : "Action2"
    },
    "stringValue" : "string02",
    "nodeValue" : null
  }, {
    "id" : "55",
    "propertyType" : {
      "id" : "55834595400",
      "name" : "Action3"
    }
]
}

输出可以是Json或csv格式。例如，在csv中： 使用与密钥相同的workorder_id

workorder_id,id_properties,stringValue_properties
99999,11,string01
99999,22,string02
99999,33,string03
.
.
.

谢谢你的帮助

Answer 1

您需要将这些空值转换为None或其他Python类型

假设您将空值切换为无

data = {'workorder_id': '99999', 'properties': [{'id': '11', 'propertyType': {'id': '55834595398', 'name': 'action'}, 'stringValue': 'string01', 'nodeValue': None}, {'id': '22', 'propertyType': {'id': '55834595419', 'name': 'Tipo'}, 'stringValue': 'string02', 'nodeValue': None}]}

def _handler(prepend: str, record: dict):
    """ relabels keys with a prepended string """
    d = {}
    for key in record.keys():
        new_key = '%s_%s' % (prepend, key)
        d.update({new_key: record[key]})
    return d

def mapper(data: dict):
    """ returns a list of dictionaries. """
    records= []
    properties = data['properties']
    for prop in properties:
        _record = {'workorder_id': data['workorder_id'], }
        
        for prop_key in prop.keys():
            prop_data = {}

            value = prop[prop_key]
            if isinstance(value, dict):
                prop_data.update(_handler(prop_key, value))
            else:
                prop_data.update({prop_key: value})

            _record.update(_handler('property', prop_data))
        records.append(_record)
    return records

输出

[{'workorder_id': '99999',
  'property_id': '11',
  'property_propertyType_id': '55834595398',
  'property_propertyType_name': 'action',
  'property_stringValue': 'string01',
  'property_nodeValue': None},
 {'workorder_id': '99999',
  'property_id': '22',
  'property_propertyType_id': '55834595419',
  'property_propertyType_name': 'Tipo',
  'property_stringValue': 'string02',
  'property_nodeValue': None}]

Answer 2

# -*- coding: utf-8 -*-

data = {
  "workorder_id" : "99999",
  "properties" : [ 
  {
    "id" : "11",
    "propertyType" : {
      "id" : "55834595398",
      "name" : "action"
    },
    "stringValue" : "string01",
    "nodeValue" : float('nan')
  }, 
  {
    "id" : "22",
    "propertyType" : {
      "id" : "55834595419",
      "name" : "Tipo"
    },
    "stringValue" : "string02",
    "nodeValue" : float('nan')
  }, 
  {
    "id" : "33",
    "propertyType" : {
      "id" : "44",
      "name" : "Action2"
    },
    "stringValue" : "string03",
    "nodeValue" : float('nan')
  }, 
  {
    "id" : "55",
    "propertyType" : {
      "id" : "55834595400",
      "name" : "Action3"
      },
    "stringValue" : "string04",
    "nodeValue" : float('nan')
  }
]
}

for item in data['properties']:
    print(data['workorder_id'], item['id'], item['propertyType']['id'], item['propertyType']['name'], item['stringValue'], item['nodeValue'])
    

#99999 11 55834595398 action string01 nan
#99999 22 55834595419 Tipo string02 nan
#99999 33 44 Action2 string03 nan
#99999 55 55834595400 Action3 string04 nan

Answer 3

使用NiFi

按此顺序：

1. EvaluateJsonPath从workorder_id（$.workorder_id）创建属性

2. 输出将其发送到$.properties.*上的拆分Json

3. splitjson的输出将其发送到evaluatejson，您将在那里提取数组

   id = $.id
   propertyType_id = $.propertyType.id
   propertyType_name = $.propertyType.name
   

现在，您的每个流都将具有以下属性：

workorder_id,id,propertyType_id,propertyType_name

1. 使用此列表使用AttributestoCSV

   工单id、id、属性类型id、属性类型名称

2. 合并内容

3. putfile（保存您的csv）