Support Questions

Find answers, ask questions, and share your expertise
Announcements
Celebrating as our community reaches 100,000 members! Thank you!

putAttribute which is generated from Nifi ExecuteScript process method

avatar
New Contributor

Hi,

I am having a requirement where user need to use execute script processor with python language to get the summation of all values from content file and summation result should add as new attribute to flow file, I am able to get the sum, But i am not getting how to pass newly created attribute to session for posting the attribute. Below is my code snippet.

import traceback
from org.apache.nifi.processors.script import ExecuteScript
from org.apache.nifi.processor.io import StreamCallback
from java.io import BufferedReader, InputStreamReader, OutputStreamWriter
class ConvertFiles(StreamCallback) :
    def __init__(self) :
        pass
    def process(self, inputStream, outputStream) :
        try :
            reader = InputStreamReader(inputStream,"UTF-8")
            bufferedReader = BufferedReader(reader)
            writer = OutputStreamWriter(outputStream,"UTF-8")
            line = bufferedReader.readLine()
            while line != None:
                total = 0
                ChangedRec = line.upper()
                writer.write(ChangedRec)
                writer.write('\n')
                a=line.split(",")
                for valu in a:
                    b=valu.strip()
                    total += int(b)
                line = bufferedReader.readLine()
            print("Summation of Records are %s ",total)
            writer.flush()
            writer.close()
            reader.close()
            bufferedReader.close()
        except :
            print "Exception in Reader:"
            print '-' * 60
            traceback.print_exc(file=sys.stdout)
            print '-' * 60
            raise
            session.transfer(flowFile, ExecuteScript.REL_FAILURE)
        finally :
            if bufferedReader is not None :
                bufferedReader.close()
            if reader is not None :
                reader.close()
flowFile = session.get()
if flowFile is not None :
    ConvertFilesData = ConvertFiles()
    session.write(flowFile, ConvertFilesData)
    flowFile = session.putAttribute(flowFile, "FileSum",ConvertFilesData.total)
    session.transfer(flowFile, ExecuteScript.REL_SUCCESS)

Now I should get Total as value and attribute name as SumResult, Could you please help me to get the issue resolved.

1 ACCEPTED SOLUTION

avatar
Master Guru
hide-solution

This problem has been solved!

Want to get a detailed solution you have to login/registered on the community

Register/Login
2 REPLIES 2

avatar
Master Guru
hide-solution

This problem has been solved!

Want to get a detailed solution you have to login/registered on the community

Register/Login

avatar
New Contributor

Thanks a lot @Matt Burgess , By adding reference self to the variable as 'self.total', It did miracle. code snippet is working absolutely fine as expected. It really helps me a lot.

import traceback
from org.apache.nifi.processors.script import ExecuteScript
from org.apache.nifi.processor.io import StreamCallback
from java.io import BufferedReader, InputStreamReader, OutputStreamWriter
class ConvertFiles(StreamCallback) :
    def __init__(self) :
        pass
    def process(self, inputStream, outputStream) :
        try :
            self.total = 0
            reader = InputStreamReader(inputStream,"UTF-8")
            bufferedReader = BufferedReader(reader)
            writer = OutputStreamWriter(outputStream,"UTF-8")
            line = bufferedReader.readLine()
            while line != None:
                ChangedRec = line.upper()
                writer.write(ChangedRec)
                writer.write('\n')
                a=line.split(",")
                for valu in a:
                    b=valu.strip()
                    self.total += int(b)
                line = bufferedReader.readLine()
            print("Summation of Records are %s ",self.total)
            writer.flush()
            writer.close()
            reader.close()
            bufferedReader.close()
        except :
            print "Exception in Reader:"
            print '-' * 60
            traceback.print_exc(file=sys.stdout)
            print '-' * 60
            raise
            session.transfer(flowFile, ExecuteScript.REL_FAILURE)
        finally :
            if bufferedReader is not None :
                bufferedReader.close()
            if reader is not None :
                reader.close()
flowFile = session.get()
if flowFile is not None :
    ConvertFilesData = ConvertFiles()
    session.write(flowFile, ConvertFilesData)
    flowFile = session.putAttribute(flowFile, "FileSum",str(ConvertFilesData.total))
    session.transfer(flowFile, ExecuteScript.REL_SUCCESS)

Snapshot Result:

88399-flowfile.png