Scripts for the NIFI API
Project description
NIFI REST API
Automates the Nifi dataflows
Dataflows can be fully automated using the Nifi API. This project taps and un-taps a dataflow by tracking the incoming/outgoing flowfiles.
More specifically, it turns on an initial processor and records the flowfiles generated by inspecting an incoming connection. After this, a middle processor is turned on. When all the flowfiles are registered in a outcoming connection, a final processor is turned on. Before finishing, it returns the pipeline to its initial state.
What is automated?
The following example represents the actions on the dataflow that get automated by the nifi_api library.
1. Consider the initial state of the dataflow:
2. Turn on the "Initial" processor and turn off the "Final" processor:
3. Turn off the "Initial" processor and record the flowfiles in the "Initial" connection, then turn on the "Middle" processor:
4. Record the flowfiles in the "Final" Connection, when they coincide with the "Initial" flowfile turn off the "Middle" processor and turn on the "Final" processor.
5. The flowfiles in the "Final" connection get consumed by the "Final" processor and the initial state of the dataflow is recovered.
Here is a recording of the Nifi UI when the tool is being executed on this dataflow:
Environment
The Nifi cluster used for testing is in the Cloudera Public Cloud and needs basic authentication credentials for accessing. The following are the environment variables required to access the cluster:
- CLOUDERA_USER=user
- CLOUDERA_PASS=password
- CLOUDERA_CLUSTER=https://<url_clustername>.cloudera.site/<clustername>
- CLOUDERA_NIFI_REST=/cdp-proxy-api/nifi-app/nifi-api/
Install
nipi_api is in pip
and can be installed by running:
pip install nifi-api
How to use
Consider the template Test_API.json
in the root folder, this is the template used in What automates?
section.
Write the data structure with the Nifi Ids (located in view configuration -> settings -> Id) of the processors and connections:
from nifi_api.environment import DataFlowIds
ids = {
"in_connection": {
"Id": "cc549c6e-0177-1000-ffff-ffffb5d2aba2",
"name": "First"
},
"out_connection": {
"Id": "51ab3b24-084f-1309-0000-00001946f2c7",
"name": "Final"
},
"in_processor": {
"Id": "36c62ad6-d606-3b04-9743-d77b6249608c",
"name": "First"
},
"middle_processor": {
"Id": "cc54862f-0177-1000-ffff-ffffe7325a20",
"name": "Middle"
},
"out_processor": {
"Id": "51ab3b1e-084f-1309-a135-aa0100d7186b",
"name": "Final"
},
}
data_ids = DataFlowIds(ids)
Instantiate and run:
from nifi_api.dataflow import DataFlow
dataflow = DataFlow(
dataflow_ids=data_ids,
delay_seconds_after_start=5,
delay_seconds_between_checks=5,
)
dataflow.run()
pipeline watching has started..
Pipeline watching has finished ...
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for nifi_rest-0.0.3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 3c6e57ba95a89c389aaa5190324829450c7376036404c9281876b697da0dc5e8 |
|
MD5 | c37a2fa8576cb2d25a1ac285d0947fc7 |
|
BLAKE2b-256 | 0616b32d032d8ed054ef2af801aa04b99f08afad81ccd107124a39aa04a7a969 |