How to run a python script with dependencies in a virtual environment in Nifi?

Question

Is there a way in Nifi to run a python script which has modules imported from a different folder, requirements specified in a pipfile and has arguments to pass?

In short, how to execute a python script which usually runs in my virtual environment using Nifi?

The end goal for me is to pick up a file using Get File and post it to API. I tried execute process, execute streamcommand processors.

Answer 1

To perform follow-on processing on the flowfile using Python, you can use the ExecuteStreamCommand or ExecuteScript/InvokeScriptedProcessor processors.

The ExecuteStreamCommand processor will run an external shell command, like python3 my_python_script.py -arg1 string -arg2 213, which can wrap custom Python code and uses STDIN to pass the existing flowfile content and STDOUT to capture the new flowfile content. Populate the Command Arguments and Command Path properties of the processor to locate your python executable and provide CLI arguments, including flowfile attributes via NiFi Expression Language. See this answer for an example.

The ExecuteScript processor runs Jython code (Python but without access to native libraries, only Python 2.7 compatibility, and some other restrictions due to JSR-223) in the same JVM as NiFi. You can process the flowfile attributes and content directly with Python code. See this answer or this answer for more details.

Answer 2

Working case.

• I have venv environment previously created through code editor in the C:\python_nifi_env\.venv folder
• I have python file with code which imports pandas library named doing_pandas_thing.py in the same C:\python_nifi_env directory
• In my NiFi UI I created "ExecuteStreamCommand" executor, where
  • Command Path: c:\python_nifi_env\.venv\Scripts\python.exe
  • Command Arguments: C:\python_nifi_env\doing_pandas_thing.py

And that's all.

# doing_pandas_thing.py contents

import numpy as np
import pandas as pd


df = pd.DataFrame(np.random.normal(0, 0.1, 1000), columns=['number'])
df.to_csv("C:\\python_nifi_env\\output\\result.csv")