I am new to MiNiFi, and started to look out for quick tutorials on how to work with MiNiFi.
Found this cloudera – article and followed the same.
But my MiNiFi is not generating any data to my NiFi instance.
I followed this cloudera – article. Got to know the process of building up a pipeline using NiFi for MiNiFi.
From the quick – start guide of MiNiFi I got know, how to set up the dataflow that has to be run by MiNiFi.
These are things I have done to build the dataflow for MiNiFi:
- First I have created a pipeline to read from a table present in my Postgres SQL database using a ExecuteSQL processor, Remote Process Group, Input Port and logAttribute Proccessor.
Pipeline to read from the PSQL table and direct it to the Input Port Note: All my processors are in running stage.
Input Port connected to a LogAttribute to receive the data from the Remote Process Group
- Second, I have downloaded the ExecuteSQL, Remote Process Group and LogAttribute pipeline as a JSON file and named it as flow.json.raw and placed it in the conf folder and started the minifi instance.
And this the error I am facing
run-minifi.bat
2024-09-02 10:49:26,738 ERROR [main] o.apache.nifi.minifi.bootstrap.Command Exception happened during MiNiFi startup
org.apache.nifi.minifi.bootstrap.configuration.ConfigurationChangeException: Unable to perform reload of received configuration change
at org.apache.nifi.minifi.bootstrap.service.MiNiFiConfigurationChangeListener.handleChange(MiNiFiConfigurationChangeListener.java:103)
at org.apache.nifi.minifi.bootstrap.command.StartRunner.regenerateFlowConfiguration(StartRunner.java:279)
at org.apache.nifi.minifi.bootstrap.command.StartRunner.start(StartRunner.java:131)
at org.apache.nifi.minifi.bootstrap.command.StartRunner.runCommand(StartRunner.java:107)
at org.apache.nifi.minifi.bootstrap.RunMiNiFi.run(RunMiNiFi.java:133)
at org.apache.nifi.minifi.bootstrap.RunMiNiFi.main(RunMiNiFi.java:150)
Caused by: java.lang.NullPointerException: Cannot invoke "org.apache.nifi.flow.VersionedProcessGroup.getIdentifier()" because "rootGroup" is null
at org.apache.nifi.minifi.commons.service.StandardFlowEnrichService.enrichFlow(StandardFlowEnrichService.java:115)
at org.apache.nifi.minifi.bootstrap.service.MiNiFiConfigurationChangeListener.handleChange(MiNiFiConfigurationChangeListener.java:92)
... 5 common frames omitted
Here is the json definition of the flow I have created.
{
"flowContents": {
"identifier": "e112dc88-89fb-3b93-ae94-558da67bb792",
"instanceIdentifier": "a3c3c70f-0191-1000-edb6-e2cf3fd3a7d4",
"name": "flow",
"comments": "",
"position": {
"x": -288,
"y": 192
},
"processGroups": [],
"remoteProcessGroups": [
{
"identifier": "704a77a9-410d-3114-a724-b37bdc242187",
"instanceIdentifier": "a37a0ea6-0191-1000-eea1-3b9e8a9dcc34",
"name": "NiFi Flow",
"comments": "",
"position": {
"x": -280,
"y": 456
},
"targetUri": "http://localhost:8088/nifi",
"targetUris": "http://localhost:8088/nifi",
"communicationsTimeout": "60 sec",
"yieldDuration": "20 sec",
"transportProtocol": "RAW",
"proxyHost": "",
"proxyUser": "",
"inputPorts": [
{
"identifier": "9221b58f-8ff5-340d-ae7b-401cc053abb0",
"instanceIdentifier": "d4493e75-0fbf-34dc-a6ad-f5d633d1c474",
"name": "FromMiNiFi",
"remoteGroupId": "704a77a9-410d-3114-a724-b37bdc242187",
"concurrentlySchedulableTaskCount": 1,
"useCompression": false,
"batchSize": {},
"componentType": "REMOTE_INPUT_PORT",
"targetId": "9d7361b0-0191-1000-2493-b894111e3ef6",
"scheduledState": "ENABLED",
"groupIdentifier": "704a77a9-410d-3114-a724-b37bdc242187"
},
{
"identifier": "cc5997a9-f62f-3764-9a10-d13e424a4008",
"instanceIdentifier": "f2b798e2-88e4-36d8-b5af-5c667f6a2360",
"name": "Copy of receiveSQLData",
"remoteGroupId": "704a77a9-410d-3114-a724-b37bdc242187",
"concurrentlySchedulableTaskCount": 1,
"useCompression": false,
"batchSize": {},
"componentType": "REMOTE_INPUT_PORT",
"targetId": "a37f55f2-0191-1000-19c7-1260366aa0d3",
"scheduledState": "ENABLED",
"groupIdentifier": "704a77a9-410d-3114-a724-b37bdc242187"
}
],
"outputPorts": [],
"componentType": "REMOTE_PROCESS_GROUP",
"groupIdentifier": "e112dc88-89fb-3b93-ae94-558da67bb792"
}
],
"processors": [
{
"identifier": "d26346ca-8682-3704-ab65-a33435b1b7d2",
"instanceIdentifier": "a38182fb-0191-1000-54f0-10cf9ad54935",
"name": "LogAttribute",
"comments": "",
"position": {
"x": 481.0553418104679,
"y": 419.9186557338876
},
"type": "org.apache.nifi.processors.standard.LogAttribute",
"bundle": {
"group": "org.apache.nifi",
"artifact": "nifi-standard-nar",
"version": "1.23.2"
},
"properties": {
"Log prefix": null,
"character-set": "UTF-8",
"Log FlowFile Properties": "true",
"Log Level": "info",
"attributes-to-ignore-regex": null,
"Attributes to Ignore": null,
"Attributes to Log": null,
"attributes-to-log-regex": ".*",
"Output Format": "Line per Attribute",
"Log Payload": "false"
},
"propertyDescriptors": {
"Log prefix": {
"name": "Log prefix",
"displayName": "Log prefix",
"identifiesControllerService": false,
"sensitive": false
},
"character-set": {
"name": "character-set",
"displayName": "Character Set",
"identifiesControllerService": false,
"sensitive": false
},
"Log FlowFile Properties": {
"name": "Log FlowFile Properties",
"displayName": "Log FlowFile Properties",
"identifiesControllerService": false,
"sensitive": false
},
"Log Level": {
"name": "Log Level",
"displayName": "Log Level",
"identifiesControllerService": false,
"sensitive": false
},
"attributes-to-ignore-regex": {
"name": "attributes-to-ignore-regex",
"displayName": "Attributes to Ignore by Regular Expression",
"identifiesControllerService": false,
"sensitive": false
},
"Attributes to Ignore": {
"name": "Attributes to Ignore",
"displayName": "Attributes to Ignore",
"identifiesControllerService": false,
"sensitive": false
},
"Attributes to Log": {
"name": "Attributes to Log",
"displayName": "Attributes to Log",
"identifiesControllerService": false,
"sensitive": false
},
"attributes-to-log-regex": {
"name": "attributes-to-log-regex",
"displayName": "Attributes to Log by Regular Expression",
"identifiesControllerService": false,
"sensitive": false
},
"Output Format": {
"name": "Output Format",
"displayName": "Output Format",
"identifiesControllerService": false,
"sensitive": false
},
"Log Payload": {
"name": "Log Payload",
"displayName": "Log Payload",
"identifiesControllerService": false,
"sensitive": false
}
},
"style": {},
"schedulingPeriod": "0 sec",
"schedulingStrategy": "TIMER_DRIVEN",
"executionNode": "ALL",
"penaltyDuration": "30 sec",
"yieldDuration": "1 sec",
"bulletinLevel": "WARN",
"runDurationMillis": 25,
"concurrentlySchedulableTaskCount": 1,
"autoTerminatedRelationships": [
"success"
],
"scheduledState": "ENABLED",
"retryCount": 10,
"retriedRelationships": [],
"backoffMechanism": "PENALIZE_FLOWFILE",
"maxBackoffPeriod": "10 mins",
"componentType": "PROCESSOR",
"groupIdentifier": "e112dc88-89fb-3b93-ae94-558da67bb792"
},
{
"identifier": "75f885a8-f584-36cf-9d90-b92a7cad5af0",
"instanceIdentifier": "a3644f7e-0191-1000-b197-019d8ad10c41",
"name": "ExecuteSQL",
"comments": "",
"position": {
"x": -136,
"y": 64
},
"type": "org.apache.nifi.processors.standard.ExecuteSQL",
"bundle": {
"group": "org.apache.nifi",
"artifact": "nifi-standard-nar",
"version": "1.23.2"
},
"properties": {
"esql-max-rows": "0",
"dbf-default-precision": "10",
"Max Wait Time": "0 seconds",
"Database Connection Pooling Service": "822e5ffc-8f36-3466-8ecc-5f319e4980c8",
"sql-post-query": null,
"esql-auto-commit": "true",
"dbf-user-logical-types": "false",
"dbf-default-scale": "0",
"sql-pre-query": null,
"compression-format": "NONE",
"esql-output-batch-size": "0",
"esql-fetch-size": "0",
"SQL select query": "select col1, col2, col3, col4, col5 from public.tjavatest;",
"dbf-normalize": "false"
},
"propertyDescriptors": {
"esql-max-rows": {
"name": "esql-max-rows",
"displayName": "Max Rows Per Flow File",
"identifiesControllerService": false,
"sensitive": false
},
"dbf-default-precision": {
"name": "dbf-default-precision",
"displayName": "Default Decimal Precision",
"identifiesControllerService": false,
"sensitive": false
},
"Max Wait Time": {
"name": "Max Wait Time",
"displayName": "Max Wait Time",
"identifiesControllerService": false,
"sensitive": false
},
"Database Connection Pooling Service": {
"name": "Database Connection Pooling Service",
"displayName": "Database Connection Pooling Service",
"identifiesControllerService": true,
"sensitive": false
},
"sql-post-query": {
"name": "sql-post-query",
"displayName": "SQL Post-Query",
"identifiesControllerService": false,
"sensitive": false
},
"esql-auto-commit": {
"name": "esql-auto-commit",
"displayName": "Set Auto Commit",
"identifiesControllerService": false,
"sensitive": false
},
"dbf-user-logical-types": {
"name": "dbf-user-logical-types",
"displayName": "Use Avro Logical Types",
"identifiesControllerService": false,
"sensitive": false
},
"dbf-default-scale": {
"name": "dbf-default-scale",
"displayName": "Default Decimal Scale",
"identifiesControllerService": false,
"sensitive": false
},
"sql-pre-query": {
"name": "sql-pre-query",
"displayName": "SQL Pre-Query",
"identifiesControllerService": false,
"sensitive": false
},
"compression-format": {
"name": "compression-format",
"displayName": "Compression Format",
"identifiesControllerService": false,
"sensitive": false
},
"esql-output-batch-size": {
"name": "esql-output-batch-size",
"displayName": "Output Batch Size",
"identifiesControllerService": false,
"sensitive": false
},
"esql-fetch-size": {
"name": "esql-fetch-size",
"displayName": "Fetch Size",
"identifiesControllerService": false,
"sensitive": false
},
"SQL select query": {
"name": "SQL select query",
"displayName": "SQL select query",
"identifiesControllerService": false,
"sensitive": false
},
"dbf-normalize": {
"name": "dbf-normalize",
"displayName": "Normalize Table/Column Names",
"identifiesControllerService": false,
"sensitive": false
}
},
"style": {},
"schedulingPeriod": "0 sec",
"schedulingStrategy": "TIMER_DRIVEN",
"executionNode": "ALL",
"penaltyDuration": "30 sec",
"yieldDuration": "1 sec",
"bulletinLevel": "WARN",
"runDurationMillis": 0,
"concurrentlySchedulableTaskCount": 1,
"autoTerminatedRelationships": [],
"scheduledState": "ENABLED",
"retryCount": 10,
"retriedRelationships": [],
"backoffMechanism": "PENALIZE_FLOWFILE",
"maxBackoffPeriod": "10 mins",
"componentType": "PROCESSOR",
"groupIdentifier": "e112dc88-89fb-3b93-ae94-558da67bb792"
}
],
"inputPorts": [],
"outputPorts": [],
"connections": [
{
"identifier": "a4c0a496-5351-3cc3-a4fb-4aadcfb93ac4",
"instanceIdentifier": "a3819773-0191-1000-47c5-0adb24a80f96",
"name": "",
"source": {
"id": "75f885a8-f584-36cf-9d90-b92a7cad5af0",
"type": "PROCESSOR",
"groupId": "e112dc88-89fb-3b93-ae94-558da67bb792",
"name": "ExecuteSQL",
"comments": "",
"instanceIdentifier": "a3644f7e-0191-1000-b197-019d8ad10c41"
},
"destination": {
"id": "d26346ca-8682-3704-ab65-a33435b1b7d2",
"type": "PROCESSOR",
"groupId": "e112dc88-89fb-3b93-ae94-558da67bb792",
"name": "LogAttribute",
"comments": "",
"instanceIdentifier": "a38182fb-0191-1000-54f0-10cf9ad54935"
},
"labelIndex": 1,
"zIndex": 0,
"selectedRelationships": [
"failure"
],
"backPressureObjectThreshold": 10000,
"backPressureDataSizeThreshold": "1 GB",
"flowFileExpiration": "0 sec",
"prioritizers": [],
"bends": [],
"loadBalanceStrategy": "DO_NOT_LOAD_BALANCE",
"partitioningAttribute": "",
"loadBalanceCompression": "DO_NOT_COMPRESS",
"componentType": "CONNECTION",
"groupIdentifier": "e112dc88-89fb-3b93-ae94-558da67bb792"
},
{
"identifier": "51ef2902-0ce4-3254-8398-0f07b666ec6b",
"instanceIdentifier": "a380e710-0191-1000-5968-000a66ac2604",
"name": "",
"source": {
"id": "75f885a8-f584-36cf-9d90-b92a7cad5af0",
"type": "PROCESSOR",
"groupId": "e112dc88-89fb-3b93-ae94-558da67bb792",
"name": "ExecuteSQL",
"comments": "",
"instanceIdentifier": "a3644f7e-0191-1000-b197-019d8ad10c41"
},
"destination": {
"id": "cc5997a9-f62f-3764-9a10-d13e424a4008",
"type": "REMOTE_INPUT_PORT",
"groupId": "704a77a9-410d-3114-a724-b37bdc242187",
"name": "Copy of receiveSQLData",
"instanceIdentifier": "f2b798e2-88e4-36d8-b5af-5c667f6a2360"
},
"labelIndex": 1,
"zIndex": 0,
"selectedRelationships": [
"success"
],
"backPressureObjectThreshold": 10000,
"backPressureDataSizeThreshold": "1 GB",
"flowFileExpiration": "0 sec",
"prioritizers": [],
"bends": [],
"loadBalanceStrategy": "DO_NOT_LOAD_BALANCE",
"partitioningAttribute": "",
"loadBalanceCompression": "DO_NOT_COMPRESS",
"componentType": "CONNECTION",
"groupIdentifier": "e112dc88-89fb-3b93-ae94-558da67bb792"
}
],
"labels": [],
"funnels": [],
"controllerServices": [],
"variables": {},
"defaultFlowFileExpiration": "0 sec",
"defaultBackPressureObjectThreshold": 10000,
"defaultBackPressureDataSizeThreshold": "1 GB",
"componentType": "PROCESS_GROUP",
"flowFileOutboundPolicy": "STREAM_WHEN_AVAILABLE",
"flowFileConcurrency": "UNBOUNDED"
},
"externalControllerServices": {
"822e5ffc-8f36-3466-8ecc-5f319e4980c8": {
"identifier": "822e5ffc-8f36-3466-8ecc-5f319e4980c8",
"name": "DBCPConnectionPool"
}
},
"parameterContexts": [],
"flowEncodingVersion": "1.0",
"parameterProviders": [],
"latest": false
}
1