Hey there!
BACKGROUND: I'm looking to setup a NiFi scheduler to run some data transformations. We have clickstream data and need to aggregate user data to find out which different statistics, such as which reports are most popular, and who are the power users.
SETUP: QueryElasticSearchHTTP processor is getting the click stream data, but the flow files are only 1 record at a time. These are being feed into an ExecuteScipt processor to concatenate the flowfile contents into a single flow file content. This file will be sent into another ExecuteScipt processor to run a python script that will produce the final data state.
PROBLEM: I'm getting a Transfer Relationship not specified error my ExecuteScript processor, that is concatenating flow files.
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from java.io import BufferedReader, InputStreamReader
from org.apache.nifi.processor.io import OutputStreamCallback
from org.apache.nifi.processor.io import InputStreamCallback
from org.apache.nifi.processor.io import StreamCallback
flowFileList = session.get(100) ##Get next 100 Rows
output = '{[' ##create what will be the new content string
ff = session.create() ##Create a new flowfile
if not flowFileList.isEmpty(): ##Check that my list isn't empty
for flowFile in flowFileList:
reader = PyInputStreamCallback() ##Create Reader
session.read(flowFile, reader) ##Readfile content
output = output + reader.getText() ##Append content to output string
# Process each FlowFile here
writer = PyOutputStreamCallback() ##Create Writer
output = output + ']}' ## Append ending to output
writer.setText(output) ## Set text that will be written to flowfile
ff = session.write(ff, writer) ##Write to flowfile
session.transfer(ff, REL_SUCCESS) ##Send flowfile to success pipe
# Define a subclass of OutputStreamCallback for use in session.write()
class PyOutputStreamCallback(OutputStreamCallback):
_text = None ##Just a holder for text
def __init__(self):
pass
def setText(self, text) :
self._text = text
def process(self, outputStream):
outputStream.write(bytearray(output.encode('utf-8'))) ##Write output to the flow file
# end class
# Define a subclass of InputStreamCallback for use in session.read()
class PyInputStreamCallback(InputStreamCallback):
_text = None##Just a holder for text
def __init__(self):
pass
def getText(self) :
return self._text
def process(self, inputStream):
reader = InputStreamReader(inputStream) ##Make Reader
bufferedReader = BufferedReader(reader) ##Make input buffer
self._text = bufferedReader.readLine() ## Read the file
# end class
