I have to know the status of data transfer job(flow inside a Process Group) whether it is completed, failed or is it running. I want to do this using nipyapi for a web application.
I have a process group in NiFi, inside which I have the NiFi flow. I am scheduling the process group using nipyapi:
nipyapi.canvas.schedule_process_group(id, True)
Now I want to monitor the status of the process group using nipyapi. By status I specifically want to know whether it’s still running, failed or completed.
Advertisement
Answer
I think I found a good solution for this problem. This is how i solved it.
So i have a mysql db which basically keeps track of all the files that are to be transferred. The database table will have 2 columns. One for the Filename (lets say is unique) and flag for whether the file has been transferred (True and False).
For Nifi Screenshot click here
We have 3 section of processors.
First: listSFTP and putMySQL
Second: getSFTP and putHDFS
Third: listHDFS and putHDFS
First Section responsible for listing the files in SFTP. It Gets all the files and adds a row into the mysql that the Filename is ‘X’ and ‘False’ for not transferred yet.
insert into NifiTest.Jobs values('${filename}', 0);
Third Section does the same thing for HDFS. It will either insert with Transferred = True or update if already a row exists with the same file name.
insert into NifiTest.Jobs values('${filename}', 1) on duplicate key update TRANSFERRED = 1;
2nd Section does nothing but Send the file to the HDFS.
Now to check when the data transfer job is finished.
You will start the entire process group together. When you query the Database and you get all Transferred = 1, that means the job is finished.
It might feel like there are some cases where it might fail but when you carefully think out all the cases you will see that it takes care of all the situations.
Let me know if i am wrong or some improvement can be made to this solution.