Nifi getting error in ExecuteScript

Question

I have been trying to extract data from nifi's ExecuteScript Processor and attach to the flowfile as attributes. I have tried many sources out there, especially the one at funnifi blog of Matt Burgess.

Following is my code

import json
import java.io
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

originalFlowFile = session.get()
text = IOUtils.toString(originalFlowFile)
log.info(text)

if(originalFlowFile != None):
    event = json.loads(text)
    if (event['true'] == 'Y'):
        flowfile = session.putAttribute(flowfile, "true", "Y")
    elif (event['src'] == 'ONE' ):
        allAttributes = { "true": "N", "src": "ONE" }
        flowfile = session.putAllAttributes(flowfile, allAttributes)
    elif (event['src'] == 'TWO' ):
        allAttributes = { "true": "N", "src": "TWO" }
        flowfile = session.putAllAttributes(flowfile, allAttributes)
    session.transfer(flowFile, REL_SUCCESS)
    session.commit()

This is in python. The flowfile data IS a JSON. Still, I am unable to parse it.

The INFO part is the output of

text = IOUtils.toString(originalFlowFile)

Any help would be greatly appreciated.

p.s. I am unfamiliar with python

Test data

{
  "true":"N",
  "src":"ONE",
  "var1":"value1",
  "var2":"value2"
}

UPDATE

My updated code, which still does not work:

import json
import java.io
from org.apache.commons.io import IOUtils

originalFlowFile = session.get()

if(originalFlowFile != None):
    inputStream = session.read(originalFlowFile)
    text = IOUtils.toString(inputStream)
    log.info(text)
    event = json.loads(text)
     if (event['true'] == 'Y'):
        flowfile = session.putAttribute(flowfile, "true", "Y")
    elif (event['src'] == 'ONE' ):
        allAttributes = { "true": "N", "src": "ONE" }
        flowfile = session.putAllAttributes(flowfile, allAttributes)
    elif (event['src'] == 'TWO' ):
        allAttributes = { "true": "N", "src": "TWO" }
        flowfile = session.putAllAttributes(flowfile, allAttributes)
    session.transfer(flowFile, REL_SUCCESS)
    session.commit()

error no json object could be decoded means you don't have a json object in flow file content. could you update your question and add: 1) a flowfile content. 2) a goal what you want to do with your script. and why you choose python when you unfamiliar with it? — daggett
I wanted to use javascript, but that is unavailable in the options! The sample data, I am adding now.... — Pritish Kamath
The goal is to check data, get the data points on the basis of which I want to create separate streams and add it as attributes. — Pritish Kamath
I published the flowfile to a kafka stream, and it is verified that the received data is in fact in JSON format. — Pritish Kamath

Bryan Bende Bryan Bende · Accepted Answer · 2017-09-08T12:59:50

Calling IOUtils.toString() on the flow file itself is likely not going to work since the flow file is not an InputStream, or Reader, or something that can be read on its own. I'm actually surprised that line doesn't produce an exception.

There are two ways you could get the content of a flow file...

The first is by getting the InputStream for a flow file from the session:

originalFlowFile = session.get();
inputStream = session.read(originalFlowFile);
text = IOUtils.toString(inputStream);

The second is by using an InputStreamCallback:

flowFile = session.read(flowFile, {inputStream ->
   // read the inputStream
} as InputStreamCallback);

Nifi getting error in ExecuteScript

1 Answers