- Subscribe to RSS Feed
- Mark Question as New
- Mark Question as Read
- Float this Question for Current User
- Bookmark
- Subscribe
- Mute
- Printer Friendly Page
putAttribute which is generated from Nifi ExecuteScript process method
- Labels:
-
Apache NiFi
Created ‎08-29-2018 12:57 PM
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
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.
Created ‎08-29-2018 05:59 PM
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
I believe your "total" variable is local to the method, so you won't be able to refer to it later. Try changing the references to "self.total", that will put the variable as a member of the class instance, so you can get at it later with ConvertFilesData.total. Also note that putAttribute() expects a String for the value of the attribute, so you'll need str(ConvertFilesData.total) there.
Created ‎08-29-2018 05:59 PM
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
I believe your "total" variable is local to the method, so you won't be able to refer to it later. Try changing the references to "self.total", that will put the variable as a member of the class instance, so you can get at it later with ConvertFilesData.total. Also note that putAttribute() expects a String for the value of the attribute, so you'll need str(ConvertFilesData.total) there.
Created on ‎08-30-2018 08:58 AM - edited ‎08-18-2019 02:47 AM
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
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:
