I am trying to create a Python script in NiFi that: Reads some attributes from an incoming flowfile Read the json content of the flowfile & extract specific fields Write attributes to outgoing flowfile Overwrite incoming flowfile with new content that is created in the script (e.g. API call that returns new json) and send it to SUCCESS relationship OR remove the old flowfile and create new with desired content What i ve done so far: 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,InputStreamCallback, OutputStreamCallback class OutputWrite(OutputStreamCallback, obj): def __init__(self): self.obj = obj def process(self, outputStream): outputStream.write(bytearray(json.dumps(self.obj).encode('utf'))) ###end class### flowfile = session.get() if flowfile != None: **#1) Get flowfile attributes** headers = { 'Accept-Encoding': 'gzip, deflate, br', 'Accept': 'application/json, text/plain, */*', 'Cache-Control': 'no-cache', 'Ocp-Apim-Trace': 'true', 'Authorization': flowfile.getAttribute('Authorization') } collection = flowfile.getAttribute('collection') dataset = flowfile.getAttribute('dataset') **#2)Get flowfile content** stream_content = session.read(flowfile) text_content = IOUtils.toString(stream_content, StandardCharsets.UTF_8) json_content = json.loads(text_content) records = json_content['result']['count'] pages = records/10000 **#3) Write flowfile attributes** flowfile = session.putAttribute(flowfile, 'collection', collection) flowfile = session.putAttribute(flowfile, 'dataset', dataset) **#API operations: output_json with desired data** output_json = {some data} **#4) Write final JSON data to output flowfile** flowfile = session.write(flowfile, OutputWrite(output_json)) session.transfer(flowfile, REL_SUCCESS) session.commit() My problem is that i can't find a way to pass a reference to the desired output_json object as an argument in the OutputStreamCallback class. Any ideas on how to resolve this or maybe a better approach? Is it maybe easier to perform all API operations in this case within the process function of the class, but then how do i get access to the incoming flowfile attributes within the process function (requires a session or a flowfile object) ? Any help much appreciated! ... View more

Hello, Trying to setup the NiFi dataflow for the HDF tutorial, interested in the Kafka part but as far as i understand i must first complete the NiFi part. However the ExecuteProcess processor gives 'Permission denied' error for the generate.sh script: Even though i ve given all permissions to the whole repo: Any ideas? ... View more

I have a use case where JSON files are read from an API, transformed to CSV and imported to Hive tables, however my flow fails at the replace text processor. Can you give some advice on the configuration of the processor or on where my approach fails? InvokeHTTP --> EvaluateJsonPath --> ReplaceText --> MergeContent --> UpdateAttribute --> PutHDFS My flow does several HTTP calls with InvokeHTTP (Each call with different ID), extracts attributes from each JSON that is returned (each JSON is unique) and then creates the csv's in the ReplaceText processor as following: ${attribute1},${attribute2},${attribute3},${attribute4},${attribute5},${attribute6},${attribute7} However after the MergeContent processor inthe merged CSV there is really a lot of duplicate data while all incoming JSONs contain unique data. ... View more

I need to read an API where my first call will return a JSON with objectIDs in this form: { objectIdFieldName: "ID", objectIds: [ 64916, 67266, 67237, 64511, .... ..] } I need to use the objectIds above to send requests for each of these IDs to the API that will return the data that i need. I was thinking a flow: GetHTTP (get response JSON) --> EvaluateJSONpath (parse only objectIds field: $.objectIds) --> ? --> InvokeHTTP (new query per ID) My problem comes after that as what i get is a sort of objectIds array in the form of: [64916, 67266, 67237, 64511,...,] How to i manage to split/parse each ID from this array in a flowfile attribute so that i can send it along with other data/headers to the InvokeHTTP processor? I thought to use SplitJson processor but i am having difficulties to understand its usage in this case. Any help much appreciated! ... View more

Trying to get files from an FTP server with ListFTP/FetchFTP, but these speicifc processors are so confusing to me. I have scheduled ListFTP to fire also with Cron and time-driven every 10 secs or so, but even though it shows a task is executed in the ListFTP processor, no flowfiles come out of it! GetFTP works fine, but i wanna implement it with List/Fetch to get only the new files in the dir. Initially i scheduled the flow with cron at 03:00 last night when i knew a new file would become available in the FTP server around 1. However this morning i saw that nothing was transfered to HDFS. So i changed scheduling to every 60 secs to test it right away and guess what, i got the new file! So i thought Cron is the problem, deleted the test-file from HDFS and scheduled the flow to run at 10:00 this morning, as expected a task was created but no flowfiles were passed to FetchFTP. Switching back to timer-driven scheduling to get the file i deleted back to HDFS from the FTP server, but this time there are no flowfiles created even for this scheduling option. What is going on here guys? ... View more

In the flow below, the goal is to take some zip archives from an FTP server, unzip them and send them to HDFS. The implication is that due to rerouting issues in this specific FTP location, i cannot use the built-in Get/List/FetchFTP processors as they fail to follow the proper rerouting. What i can do is use a command line utility from the Nifi server that can handle rerouting, and indeed ncftp in this case does the trick. So my plan is to use ExecuteProcess to run the bash script: ncftpget -f login.txt /home/user/test /path/to/remote/*.zip > /dev/null 2>&1 ls /home/user/test | grep ".zip" The first line gets the wanted zip archives from the FTP server while redirects all output/error streams, as we want to parse only the output of the second line which lists the contents of the specified directory and parses the ones with 'zip' extension. What i am trying to do is to recreate the proper filenames between ExtractText --> UpdateAttribute and pass them to FetchFile --> UnpackContent --> PutHDFS. So the output of the ExecuteProcess is something like: file1_timestamp.zip file2_timestamp.zip file3_timestamp.zip ..... file100_timestamp.zip Next, ExtractText processor with an added property 'filename': '\w+.zip' looks for this regex in the flowfile content and outputs a flowfile with new attributes filename1,filename2...filename100 for each match. Subsequently, UpdateAttribute specifies the local path the zip archives have been placed from our bash script ('/home/user/test' in this case), as well as the proper filename so that ${path}/${filename} are passed to the rest of the flow for fetching, unpacking and finally putting to HDFS. The problem i have is that only the first match is passed to the rest of the flow as only this match corresponds to the 'filename' attribute. The other filenames are parsed according to the ExtractText processor to the attributes 'filename.2', 'filename.3'... 'filename.100'. I would like to find a way to update the attributes passed to FetchFile with some kind of incremental counter. I tried to configure the FetchFile processor with File to Fetch property as ${path}/${filename: nextInt()} but this just looks for 'file_timestamp.zip#' filenames in the specified path that ofc are not there. ... View more

I have a directory with zip archives in the local filesystem of the Nifi server, and i would like to create a flow that unzips these archives with a bash script and then puts them in HDFS. The problem i have is that i cannot direct the output of the bash script in a correct way to the PutHDFS processor so that it parses the unzipped files. 1) With the use of ExecuteStreamCommand processor i have 2 options for the outgoing flow, the original relationship that contains the initial zipped archive and the outputstream relationship which it should be what i am looking for but it transfers only an empty file with the same name with the original. How should be this processor be configured when it runs a bash script/command to correctly contain the files produced from this script/command? 2) With the use of ExecuteProcess processor, where there is only a success/failure relationship and also this does not help to pass the outgoing flow as input of the PutHDFS processor to move the unzipped files to HDFS. Any help would be greately appreciated! ... View more