How to access the nested fields in a json using expression language in Nifi?

janvit04 — Wed, 26 Apr 2023 06:12:44 GMT

Hi,
I have an input json as below:

[ { "orderId": "1234567890", "orderName": "Test1", "orderItems": [ { "orderItemId": "01", "orderItemName": "test1-OrderItem1" }, { "orderItemId": "02", "orderItemName": "test1-OrderItem2" } ] }, { "orderId": "12235", "orderName": "Test2", "orderItems": [ { "orderItemId": "01", "orderItemName": "test2-OrderItem1" } ] }, { "orderId": "12236", "orderName": "Test3", "orderItems": [ { "orderItemId": "01", "orderItemName": "test3-OrderItem1" } ] } ]

I want to convert it into text form where we want the order details and then item details for each map in the list.
Here is the desired output:

1234567890 Test1 01 test1-OrderItem1 02 test1-OrderItem2 12235 Test2 01 test2-OrderItem1 12236 Test3 01 test3-OrderItem1

This is json to csv conversion, and I am using the ConvertRecord processor and below are the reader and writer configured:
1. Record Reader - JsonTreeReader
2. Record Writer - FreeFormTextRecordSetWriter

For the FreeFormTextRecordSetWriter controller service in the text property, I added the key names from the input json like this ${keyName} and got the values but in the next line, I wanted orderItemId from all the maps in the orderItems list.

How can I access the values from orderItems list?

Thanks in advance!

Re: How to access the nested fields in a json using expression language in Nifi?

joseomjr — Sat, 16 Sep 2023 23:35:03 GMT

I would do this with a Groovy based InvokeScriptedProcessor

Using this code:

import groovy.json.JsonOutput import groovy.json.JsonSlurper import java.nio.charset.StandardCharsets import org.apache.commons.io.IOUtils class GroovyProcessor implements Processor { PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder() .name("BATCH_SIZE") .displayName("Batch Size") .description("The number of incoming FlowFiles to process in a single execution of this processor.") .required(true) .defaultValue("1000") .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) .build() Relationship REL_SUCCESS = new Relationship.Builder() .name("success") .description('FlowFiles that were successfully processed are routed here') .build() Relationship REL_FAILURE = new Relationship.Builder() .name("failure") .description('FlowFiles that were not successfully processed are routed here') .build() ComponentLog log void initialize(ProcessorInitializationContext context) { log = context.logger } Set<Relationship> getRelationships() { return [REL_FAILURE, REL_SUCCESS] as Set } Collection<ValidationResult> validate(ValidationContext context) { null } PropertyDescriptor getPropertyDescriptor(String name) { null } void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue) { } List<PropertyDescriptor> getPropertyDescriptors() { Collections.unmodifiableList([BATCH_SIZE]) as List<PropertyDescriptor> } String getIdentifier() { null } JsonSlurper jsonSlurper = new JsonSlurper() JsonOutput jsonOutput = new JsonOutput() void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { ProcessSession session = sessionFactory.createSession() try { List<FlowFile> flowFiles = session.get(context.getProperty(BATCH_SIZE).asInteger()) if (!flowFiles) return flowFiles.each { flowFile -> List data = null session.read(flowFile, { inputStream -> data = jsonSlurper.parseText(IOUtils.toString(inputStream, StandardCharsets.UTF_8)) } as InputStreamCallback) List outputData = [] data.each { order -> outputData.add("${order.orderId} ${order.orderName}") order.orderItems.each { orderItem -> outputData.add("${orderItem.orderItemId} ${orderItem.orderItemName}") } } FlowFile newFlowFile = session.create() newFlowFile = session.write(newFlowFile, { outputStream -> outputStream.write(outputData.join('\n').getBytes(StandardCharsets.UTF_8)) } as OutputStreamCallback) session.transfer(newFlowFile, REL_SUCCESS) session.remove(flowFile) } session.commit() } catch (final Throwable t) { log.error('{} failed to process due to {}; rolling back session', [this, t] as Object[]) session.rollback(true) throw t } } } processor = new GroovyProcessor()

Don't let all that code scare you when the part that's doing the formatting is only these lines:

This is the generated output:

question How to access the nested fields in a json using expression language in Nifi? in Support Questions

How to access the nested fields in a json using expression language in Nifi?

Re: How to access the nested fields in a json using expression language in Nifi?