Member since
07-30-2019
1925
Posts
1152
Kudos Received
525
Solutions
My Accepted Solutions
| Title | Views | Posted |
|---|---|---|
| 45 | 01-21-2021 05:40 AM | |
| 104 | 01-05-2021 11:19 AM | |
| 155 | 01-04-2021 08:23 AM | |
| 221 | 12-28-2020 10:33 AM | |
| 155 | 12-24-2020 07:35 AM |
01-21-2021
07:06 AM
@Lallagreta Make sure you do not have any line returns in the values for your dynamic properties added in the UpdateAttribute processor. When you click on the value field for each property you should not see a line "2". For example: Above would result in the value assigned to the FlowFile Attribute having a line return. If this is the case, edit the properties value(s) to remove the line returns so you only see one line (1). Hope this helps, Matt
... View more
01-21-2021
06:55 AM
@pal_lee A FlowFile represents each object moving through the system and for each one, NiFi keeps track of a map of key/value pair attribute strings and its associated content of zero or more bytes. A FlowFile consist of two parts (FlowFile Attributes/Metadata and FlowFile content). NiFi overview. I assume what you are really asking here is how to programmatically construct a NiFi dataflow via rest-api calls rather then drag and drop to the canvas via the NiFi UI. Probably the easiest way to become familiar with how to do this is through the use of the developer tools available in your browser. Each request made while constructing a dataflow via the NiFi UI would be captured by the developer tools and you can right click on those calls to get the curl command that was actually executed against the ret-api endpoint to initiate that change on the canvas. These work as great examples for how to add components and establish connections between those components. The NiFI rest-api docs are also a good source of reference but really do not cover down to the specific nuances of every component. Hope this helps, Matt
... View more
01-21-2021
05:40 AM
2 Kudos
@Siddo The current strategy you are using is the best option with a use a case where the client is sending/pushing data to listeners across your NiFi cluster nodes. Whenever you have a client that is pushing data to NiFi, this setup avoids as your mentioned having a single point of failure. If a load balancer can't be used, It becomes the responsibility of the client to detect delivery problems and switch to delivering to a different node. Load balancing within NiFi's dataflows is the best option when the dataflow is consuming from a source system. Some data consumption methods are not cluster friendly (for example FTP). This is because every node in a NiFi cluster executes the same flow.xml.gz. If you had for example the listSFTP/GetSFTP processors running on every node, you would have data duplication and potentially issues as every node tried to consume the same data. So in this scenario you would configure the processor to execute on the primary node only and then use LB connections to immediately redistribute those FlowFiles across your cluster before doing further processing. This is why we created the List and Fetch processor pairs. These are typically non cluster friendly type processors. So a ListSFTP produces FlowFiles with zero content and only attributes with details on where to fetch a specific FlowFiles content. Those 0 bytes FlowFiles quickly Load balance across the cluster where the FetchSFTP processor would fetch the actual content for the FlowFile specific data file and insert it into the FlowFile. This type of setup also avoid single point of failure since loss of the currently elected primary node (where the data lister/consumer is running) would result in a new node being elected as the new Primary node. That new primary node reads state from the cluster state provider and begins listing where the previous elected node's list processor stopped. So you can see that each use case has very specific benefits/use cases. Another scenario may be that even with an external F5 LB, you may find one node in your cluster ends up with a larger burden of work load (maybe one node ends up with bulk of larger data files. That data can be redistribute on connections were such single node bottle necks occur to re-balance the load at that point in a dataflow. So at times a combination may make sense as well, but I would not just apply this strategy unless needed since it adds to network usage. NiFi's internal LB connections can also be used to move all data to a single node for some use case. Let's say there is a batch of data spread out across multiple NiFi nodes that you want to merge in to a single FlowFile. NiFi nodes each work on only the FlowFiles on their own node. But using LB connection in specific spots on your flow would allow you to move all like data to the sam node before a merge type processor. Hope this helps, Matt
... View more
01-15-2021
07:11 AM
MiNiFi offers CPP version that is well suited for Windows event log ingestion.
... View more
01-15-2021
07:07 AM
@dzbeda Just to add to this, MiNiFi offers a C++ agent. There are many users out there using MiNiFi CPP collecting Wiindows event logs and forwarding them to nifi via invokehttp (on MiNiFi CPP) to listenhttp (on NiFi). Thanks, Matt
... View more
01-15-2021
06:24 AM
@Melissa It may be helpful if you can share screenshots of your controller services (JsonTreeReader and JsonRecordSetWriter complete configurations) and the complete exception written to the nifi-app.log.
... View more
01-15-2021
06:06 AM
1 Kudo
@Lyoung The NiFi client (NiFi or MiNiFi instance running the Remote Process Group (RPG)) has not control over the connection with the server (NiFi configured with Remote input or Output ports). The RPG is provided with a http or https address of one or more target NiFi nodes in a NiFi cluster). A background thread connects to that target NiFi to fetch Site-To-Site (S2S) details. If the target is https enabled, a mutual TLS handshake will happen. This means the client must have a keystore and truststore configured in the nifi.properties (NiFi) or config.yaml (MiNiFi) that can successfully be used to mutually authenticate with the target NiFi server. The server side NiFi must have the properties you listed configured: nifi.remote.input.host=<must be set to hostname of NiFi on which you are configuring this property. This is the hostname returned to client in the S2S details. Be careful that what ever you set heer does not resolve to localhost.> nifi.remote.input.secure=false (this tells client if connection is secure or unsecure. If false, the "nifi.web.http.port" property must be set and the URL used in the RPG must be "http://<target nifi>:<http port>/nifii". If set to true, the "nifi.web.https.port" property must be set and the URL used in the RPG must be "https://<target nifi>:<https port>/nifii" nifi.remote.input.socket.port=<This is the RAW port that will be used to actually send or receive the FlowFiles from remote Input or Output ports on target NiFi node(s). If this property is not set on the target NiFi node(s), RAW transport protocol will not be supported. (S2S details are always fetched over HTTP)> nifi.remote.input.http.enabled=true. <This properties states whether the "http" transport protocol can be used for sending the FlowFiles.> nifi.remote.input.http.transaction.ttl=30 sec nifi.remote.contents.cache.expiration=30 secs Based on the log output shared it sounds like above properties were not set on the Target NiFi node(s). Did you set them on client NiFi (NiFi actually running the RPG)? In addition to the Target NiFi S2S details above for each target NiFi node being returned to client, the details will also include the FlowFile load on each node, Remote input ports that client has been authorized to use, and Remote Output ports that the client has been authorized to use. If the target server side NiFi node(s) are unsecured then there will be no authorization set for ports, all clients would have access to all remote input/output ports. Also keep in mind that any changes to NiFi's/MiNiFi's configuration files would require a restart of the service before they would be applied. Aside from above, I would need to see screenshots and nifi.properties/config.yaml configs of both your client and server side of this S2S connection to help further. Hope this helps, Matt
... View more
01-15-2021
05:15 AM
@Fierymech If you clear all the FlowFiles out of you test dataflow, stop all processors, and start on your GetFile processor, how many FlowFiles get queued on the success connection out of the GetFile processor? How many "out" does it show on the GetFile stats?
... View more
01-13-2021
11:42 AM
1 Kudo
@Fierymech I am not able to reproduce the 1file in and 7 files out that you described. Are you sure it is only 1 FlowFile in? As far as your use case goes, perhaps you use RouteText first to produce the new FlowFile with all the lines containing the data you want (This includes the leading "*" and trailing "yes yes ..." strings). To do this you would use the following regex and a "matching strategy" of "contains regular expression": \d{1,2}\s\w{1,4}\-\w{1,4}\-(\w{1,2}|\w{0})\s\d{1,5}\s\d{1,2}\-\w{1,3}\s\d{1,2}\:\d{1,2}\:\d{1,2} Then pass that new FlowFile to a ReplaceText processor which can trim off the leading whitespaces and "*" and any trailing whitespace characters and additional text. This ReplaceText would be configured as follows: And use Search Value of which contains 3 capture groups: (.*?)(\d{1,2}\s\w{1,4}\-\w{1,4}\-(\w{1,2}|\w{0})\s\d{1,5}\s\d{1,2}\-\w{1,3}\s\d{1,2}\:\d{1,2}\:\d{1,2})(.*?)$ The processor replaces text line-by-line with only the second capture group. This worked for me to get the end result of: Which is what i believe you are looking for in the resulting FlowFile's content. Hope this helps, Matt
... View more
01-13-2021
11:03 AM
1 Kudo
@Fierymech Here is the raw XML for the template: <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<template encoding-version="1.3">
<description></description>
<groupId>658f3a7f-0171-1000-0000-00007706d23d</groupId>
<name>RouteText-Example</name>
<snippet>
<connections>
<id>44a4a771-a46f-3b74-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
<backPressureObjectThreshold>10000</backPressureObjectThreshold>
<bends>
<x>0.0</x>
<y>408.0</y>
</bends>
<destination>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>e81dcc20-cb6f-3466-0000-000000000000</id>
<type>PROCESSOR</type>
</destination>
<flowFileExpiration>0 sec</flowFileExpiration>
<labelIndex>1</labelIndex>
<loadBalanceCompression>DO_NOT_COMPRESS</loadBalanceCompression>
<loadBalancePartitionAttribute></loadBalancePartitionAttribute>
<loadBalanceStatus>LOAD_BALANCE_NOT_CONFIGURED</loadBalanceStatus>
<loadBalanceStrategy>DO_NOT_LOAD_BALANCE</loadBalanceStrategy>
<name></name>
<selectedRelationships>original</selectedRelationships>
<source>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>696c796c-aa86-3e71-0000-000000000000</id>
<type>PROCESSOR</type>
</source>
<zIndex>0</zIndex>
</connections>
<connections>
<id>7bcc4c55-39cf-3b4c-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
<backPressureObjectThreshold>10000</backPressureObjectThreshold>
<destination>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>696c796c-aa86-3e71-0000-000000000000</id>
<type>PROCESSOR</type>
</destination>
<flowFileExpiration>0 sec</flowFileExpiration>
<labelIndex>1</labelIndex>
<loadBalanceCompression>DO_NOT_COMPRESS</loadBalanceCompression>
<loadBalancePartitionAttribute></loadBalancePartitionAttribute>
<loadBalanceStatus>LOAD_BALANCE_NOT_CONFIGURED</loadBalanceStatus>
<loadBalanceStrategy>DO_NOT_LOAD_BALANCE</loadBalanceStrategy>
<name></name>
<selectedRelationships>success</selectedRelationships>
<source>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>7386cd31-1cfd-3e04-0000-000000000000</id>
<type>PROCESSOR</type>
</source>
<zIndex>0</zIndex>
</connections>
<connections>
<id>dde64489-3d09-3e4f-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
<backPressureObjectThreshold>10000</backPressureObjectThreshold>
<bends>
<x>480.0</x>
<y>408.0</y>
</bends>
<destination>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>e81dcc20-cb6f-3466-0000-000000000000</id>
<type>PROCESSOR</type>
</destination>
<flowFileExpiration>0 sec</flowFileExpiration>
<labelIndex>1</labelIndex>
<loadBalanceCompression>DO_NOT_COMPRESS</loadBalanceCompression>
<loadBalancePartitionAttribute></loadBalancePartitionAttribute>
<loadBalanceStatus>LOAD_BALANCE_NOT_CONFIGURED</loadBalanceStatus>
<loadBalanceStrategy>DO_NOT_LOAD_BALANCE</loadBalanceStrategy>
<name></name>
<selectedRelationships>unmatched</selectedRelationships>
<source>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>696c796c-aa86-3e71-0000-000000000000</id>
<type>PROCESSOR</type>
</source>
<zIndex>0</zIndex>
</connections>
<connections>
<id>f2b4b8ba-09e7-3125-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
<backPressureObjectThreshold>10000</backPressureObjectThreshold>
<bends>
<x>240.0</x>
<y>408.0</y>
</bends>
<destination>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>e81dcc20-cb6f-3466-0000-000000000000</id>
<type>PROCESSOR</type>
</destination>
<flowFileExpiration>0 sec</flowFileExpiration>
<labelIndex>1</labelIndex>
<loadBalanceCompression>DO_NOT_COMPRESS</loadBalanceCompression>
<loadBalancePartitionAttribute></loadBalancePartitionAttribute>
<loadBalanceStatus>LOAD_BALANCE_NOT_CONFIGURED</loadBalanceStatus>
<loadBalanceStrategy>DO_NOT_LOAD_BALANCE</loadBalanceStrategy>
<name></name>
<selectedRelationships>matched</selectedRelationships>
<source>
<groupId>2edcec51-fc4e-38cf-0000-000000000000</groupId>
<id>696c796c-aa86-3e71-0000-000000000000</id>
<type>PROCESSOR</type>
</source>
<zIndex>0</zIndex>
</connections>
<processors>
<id>19e6484f-37ca-3639-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<position>
<x>736.0</x>
<y>0.0</y>
</position>
<bundle>
<artifact>nifi-standard-nar</artifact>
<group>org.apache.nifi</group>
<version>1.12.1.3.5.2.0-99</version>
</bundle>
<config>
<bulletinLevel>WARN</bulletinLevel>
<comments></comments>
<concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
<descriptors>
<entry>
<key>Routing Strategy</key>
<value>
<name>Routing Strategy</name>
</value>
</entry>
<entry>
<key>Matching Strategy</key>
<value>
<name>Matching Strategy</name>
</value>
</entry>
<entry>
<key>Character Set</key>
<value>
<name>Character Set</name>
</value>
</entry>
<entry>
<key>Ignore Leading/Trailing Whitespace</key>
<value>
<name>Ignore Leading/Trailing Whitespace</name>
</value>
</entry>
<entry>
<key>Ignore Case</key>
<value>
<name>Ignore Case</name>
</value>
</entry>
<entry>
<key>Grouping Regular Expression</key>
<value>
<name>Grouping Regular Expression</name>
</value>
</entry>
<entry>
<key>lines</key>
<value>
<name>lines</name>
</value>
</entry>
<entry>
<key>matched</key>
<value>
<name>matched</name>
</value>
</entry>
</descriptors>
<executionNode>ALL</executionNode>
<lossTolerant>false</lossTolerant>
<penaltyDuration>30 sec</penaltyDuration>
<properties>
<entry>
<key>Routing Strategy</key>
<value>Route to each matching Property Name</value>
</entry>
<entry>
<key>Matching Strategy</key>
<value>Contains Regular Expression</value>
</entry>
<entry>
<key>Character Set</key>
<value>UTF-8</value>
</entry>
<entry>
<key>Ignore Leading/Trailing Whitespace</key>
<value>true</value>
</entry>
<entry>
<key>Ignore Case</key>
<value>false</value>
</entry>
<entry>
<key>Grouping Regular Expression</key>
</entry>
<entry>
<key>lines</key>
<value>\d{0,2}\sabc-\w{0,2}\d{0,2}-\d{0,2}\w{0,2}\s\d{0,6}\s\d{0,2}-\w{0,3}\s\d{0,2}\:\d{0,2}\:\d{0,2}</value>
</entry>
<entry>
<key>matched</key>
<value>(\s|[ \t]|\*)\d{1,2}\s\w{1,4}\-\w{1,4}\-(\w{1,2}|\w{0})\s\d{1,5}\s\d{1,2}\-\w{1,3}\s\d{1,2}\:\d{1,2}\:\d{1,2}</value>
</entry>
</properties>
<runDurationMillis>0</runDurationMillis>
<schedulingPeriod>0 sec</schedulingPeriod>
<schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
<yieldDuration>1 sec</yieldDuration>
</config>
<executionNodeRestricted>false</executionNodeRestricted>
<name>RouteText</name>
<relationships>
<autoTerminate>false</autoTerminate>
<name>lines</name>
</relationships>
<relationships>
<autoTerminate>false</autoTerminate>
<name>matched</name>
</relationships>
<relationships>
<autoTerminate>false</autoTerminate>
<name>original</name>
</relationships>
<relationships>
<autoTerminate>false</autoTerminate>
<name>unmatched</name>
</relationships>
<state>STOPPED</state>
<style/>
<type>org.apache.nifi.processors.standard.RouteText</type>
</processors>
<processors>
<id>696c796c-aa86-3e71-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<position>
<x>64.0</x>
<y>248.0</y>
</position>
<bundle>
<artifact>nifi-standard-nar</artifact>
<group>org.apache.nifi</group>
<version>1.12.1.3.5.2.0-99</version>
</bundle>
<config>
<bulletinLevel>WARN</bulletinLevel>
<comments></comments>
<concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
<descriptors>
<entry>
<key>Routing Strategy</key>
<value>
<name>Routing Strategy</name>
</value>
</entry>
<entry>
<key>Matching Strategy</key>
<value>
<name>Matching Strategy</name>
</value>
</entry>
<entry>
<key>Character Set</key>
<value>
<name>Character Set</name>
</value>
</entry>
<entry>
<key>Ignore Leading/Trailing Whitespace</key>
<value>
<name>Ignore Leading/Trailing Whitespace</name>
</value>
</entry>
<entry>
<key>Ignore Case</key>
<value>
<name>Ignore Case</name>
</value>
</entry>
<entry>
<key>Grouping Regular Expression</key>
<value>
<name>Grouping Regular Expression</name>
</value>
</entry>
<entry>
<key>matched</key>
<value>
<name>matched</name>
</value>
</entry>
</descriptors>
<executionNode>ALL</executionNode>
<lossTolerant>false</lossTolerant>
<penaltyDuration>30 sec</penaltyDuration>
<properties>
<entry>
<key>Routing Strategy</key>
<value>Route to each matching Property Name</value>
</entry>
<entry>
<key>Matching Strategy</key>
<value>Matches Regular Expression</value>
</entry>
<entry>
<key>Character Set</key>
<value>UTF-8</value>
</entry>
<entry>
<key>Ignore Leading/Trailing Whitespace</key>
<value>true</value>
</entry>
<entry>
<key>Ignore Case</key>
<value>false</value>
</entry>
<entry>
<key>Grouping Regular Expression</key>
</entry>
<entry>
<key>matched</key>
<value>(\s|[ \t]|\*)\d{1,2}\s\w{1,4}\-\w{1,4}\-(\w{1,2}|\w{0})\s\d{1,5}\s\d{1,2}\-\w{1,3}\s\d{1,2}\:\d{1,2}\:\d{1,2}</value>
</entry>
</properties>
<runDurationMillis>0</runDurationMillis>
<schedulingPeriod>0 sec</schedulingPeriod>
<schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
<yieldDuration>1 sec</yieldDuration>
</config>
<executionNodeRestricted>false</executionNodeRestricted>
<name>RouteText</name>
<relationships>
<autoTerminate>false</autoTerminate>
<name>matched</name>
</relationships>
<relationships>
<autoTerminate>false</autoTerminate>
<name>original</name>
</relationships>
<relationships>
<autoTerminate>false</autoTerminate>
<name>unmatched</name>
</relationships>
<state>STOPPED</state>
<style/>
<type>org.apache.nifi.processors.standard.RouteText</type>
</processors>
<processors>
<id>7386cd31-1cfd-3e04-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<position>
<x>56.0</x>
<y>40.0</y>
</position>
<bundle>
<artifact>nifi-standard-nar</artifact>
<group>org.apache.nifi</group>
<version>1.12.1.3.5.2.0-99</version>
</bundle>
<config>
<bulletinLevel>WARN</bulletinLevel>
<comments></comments>
<concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
<descriptors>
<entry>
<key>File Size</key>
<value>
<name>File Size</name>
</value>
</entry>
<entry>
<key>Batch Size</key>
<value>
<name>Batch Size</name>
</value>
</entry>
<entry>
<key>Data Format</key>
<value>
<name>Data Format</name>
</value>
</entry>
<entry>
<key>Unique FlowFiles</key>
<value>
<name>Unique FlowFiles</name>
</value>
</entry>
<entry>
<key>generate-ff-custom-text</key>
<value>
<name>generate-ff-custom-text</name>
</value>
</entry>
<entry>
<key>character-set</key>
<value>
<name>character-set</name>
</value>
</entry>
<entry>
<key>mime-type</key>
<value>
<name>mime-type</name>
</value>
</entry>
</descriptors>
<executionNode>PRIMARY</executionNode>
<lossTolerant>false</lossTolerant>
<penaltyDuration>30 sec</penaltyDuration>
<properties>
<entry>
<key>File Size</key>
<value>0B</value>
</entry>
<entry>
<key>Batch Size</key>
<value>1</value>
</entry>
<entry>
<key>Data Format</key>
<value>Text</value>
</entry>
<entry>
<key>Unique FlowFiles</key>
<value>false</value>
</entry>
<entry>
<key>generate-ff-custom-text</key>
<value> Capable Qd Tx
# Name S/N Since Mas/Sla bytes
-- --------- ----- --------------- --- --- ---------
0 hub-lr1-0 35189 20-Dec 03:43:54
1 lr2-27-27 35209 20-Dec 03:43:54
2 dt27-kcd- 35185 20-Dec 03:43:54
* 3 rr1-2627- 34748 20-Dec 03:43:54 yes yes 0
4 hub-rr2-g 34609 20-Dec 03:43:54
5 hub-lr2-0 34686 20-Dec 03:43:54
6 hub-lr1-0 34631 20-Dec 03:43:54
7 hub-rr3-g 34692 20-Dec 03:43:54
8 hub-rr3-g 34568 20-Dec 03:43:54
9 hub-rr2-g 35203 20-Dec 03:43:54
10 hub-rr2-g 35200 20-Dec 03:43:54
11 hub-lr1-0 35205 20-Dec 03:43:54
12 hub-rr1-0 34394 20-Dec 03:43:54
13 hub-rr3-g 35191 20-Dec 03:43:54
14 hub-lr2-0 35196 20-Dec 03:43:54
15 hub-lr1-0 35214 20-Dec 03:43:54
16 hub-rr1-0 34577 20-Dec 03:43:54
*17 hub-rr3-g 35217 20-Dec 03:43:56
Logs for Radio IP xx.xx.xx.xx
telnet> Trying xx.xx.xx.xx...
Logs for Radio IP xx.xx.xx.xx
telnet> Trying xx.xx.xx.xx...
Connected to xx.xx.xx.xx.
Escape character is '^]'.
Capable Qd Tx
# Name S/N Since Mas/Sla bytes
-- --------- ----- --------------- --- --- ---------
0 hub-lr1-0 35189 20-Dec 03:43:54
1 hub-rr2-g 35209 20-Dec 03:43:54
2 hub-rr1-0 35185 20-Dec 03:43:54
3 hub-rr1-0 34748 20-Dec 03:43:54
4 hub-rr2-g 34609 20-Dec 03:43:54
5 hub-lr2-0 34686 20-Dec 03:43:54
6 hub-lr1-0 34631 20-Dec 03:43:54
7 hub-rr3-g 34692 20-Dec 03:43:54
8 hub-rr3-g 34568 20-Dec 03:43:54
9 hub-rr2-g 35203 20-Dec 03:43:54
10 hub-rr2-g 35200 20-Dec 03:43:54
11 hub-lr1-0 35205 20-Dec 03:43:54
12 hub-rr1-0 34394 20-Dec 03:43:54
13 hub-rr3-g 35191 20-Dec 03:43:54
14 hub-lr2-0 35196 20-Dec 03:43:54
15 hub-lr1-0 35214 20-Dec 03:43:54
16 hub-rr1-0 34577 20-Dec 03:43:54
*17 hub-rr3-g 35217 20-Dec 03:43:54 yes yes 0 </value>
</entry>
<entry>
<key>character-set</key>
<value>UTF-8</value>
</entry>
<entry>
<key>mime-type</key>
</entry>
</properties>
<runDurationMillis>0</runDurationMillis>
<schedulingPeriod>60 sec</schedulingPeriod>
<schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
<yieldDuration>1 sec</yieldDuration>
</config>
<executionNodeRestricted>false</executionNodeRestricted>
<name>GenerateFlowFile</name>
<relationships>
<autoTerminate>false</autoTerminate>
<name>success</name>
</relationships>
<state>STOPPED</state>
<style/>
<type>org.apache.nifi.processors.standard.GenerateFlowFile</type>
</processors>
<processors>
<id>e81dcc20-cb6f-3466-0000-000000000000</id>
<parentGroupId>2edcec51-fc4e-38cf-0000-000000000000</parentGroupId>
<position>
<x>64.0</x>
<y>440.0</y>
</position>
<bundle>
<artifact>nifi-update-attribute-nar</artifact>
<group>org.apache.nifi</group>
<version>1.12.1.3.5.2.0-99</version>
</bundle>
<config>
<bulletinLevel>WARN</bulletinLevel>
<comments></comments>
<concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
<descriptors>
<entry>
<key>Delete Attributes Expression</key>
<value>
<name>Delete Attributes Expression</name>
</value>
</entry>
<entry>
<key>Store State</key>
<value>
<name>Store State</name>
</value>
</entry>
<entry>
<key>Stateful Variables Initial Value</key>
<value>
<name>Stateful Variables Initial Value</name>
</value>
</entry>
<entry>
<key>canonical-value-lookup-cache-size</key>
<value>
<name>canonical-value-lookup-cache-size</name>
</value>
</entry>
</descriptors>
<executionNode>ALL</executionNode>
<lossTolerant>false</lossTolerant>
<penaltyDuration>30 sec</penaltyDuration>
<properties>
<entry>
<key>Delete Attributes Expression</key>
</entry>
<entry>
<key>Store State</key>
<value>Do not store state</value>
</entry>
<entry>
<key>Stateful Variables Initial Value</key>
</entry>
<entry>
<key>canonical-value-lookup-cache-size</key>
<value>100</value>
</entry>
</properties>
<runDurationMillis>0</runDurationMillis>
<schedulingPeriod>0 sec</schedulingPeriod>
<schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
<yieldDuration>1 sec</yieldDuration>
</config>
<executionNodeRestricted>false</executionNodeRestricted>
<name>UpdateAttribute</name>
<relationships>
<autoTerminate>false</autoTerminate>
<name>success</name>
</relationships>
<state>STOPPED</state>
<style/>
<type>org.apache.nifi.processors.attributes.UpdateAttribute</type>
</processors>
</snippet>
<timestamp>01/13/2021 16:41:18 UTC</timestamp>
</template> Save this entire xml snippet to a file with the .xml extension and import it as a template in your NiFi. Hope this helps, Matt
... View more
01-13-2021
08:46 AM
Blocks my attachment (sorry)
... View more
01-13-2021
08:45 AM
1 Kudo
@Fierymech RouteText does not modify the content of the lines. It only routes lines to different produced new FlowFiles. The content of those lines remains unchanged. The RouteText processor also does nothing with capture groups, so the entire regex is going to be evaluated against each line. I took the entire content from your "https://regex101.com/r/pdo6Ca/1" and the entire new regex from same and ran this flow. I produced only one source FlowFile as you can see processor in = 1 You can see it routed "out" three FlowFiles. One to the connection with the "matched" relationship which contains only one line since only one line matched the entire regex. (regex101 is taking in to account your capture groups) If you change: And run same test again, you will see a few more lines match (those with the additional "yes yes ..." in the lines). I attached template i used which you can import to your NiFi. Community does not support .xml files so i changed extension to .txt. You will need to change extension back to .xml before you can import the template in to your NiFi. Hope this helps.
... View more
01-13-2021
06:04 AM
1 Kudo
@Fierymech It may be helpful if you shared your RouteText processor configuration. Correct me if I am wrong, but you are looking to have all lines (minus the header lines) placed in a new FlowFile by themselves. Using you example data and the regex you provided. wwwwww aa cc
# Name foo Since ddd/www dddd
-- --------- ----- --------------- --- --- ---------
0 abc-lr1-0 35189 20-Dec 03:43:54
1 abc-rr2-g 35209 20-Dec 03:43:54
* 2 abc-rr1-0 35185 20-Dec 03:43:54
* 15 abc-lr2-0 34686 20-Dec 03:43:54
16 abc-lr1-0 34631 20-Dec 03:43:54 The above would result in a FlowFile with only lines 0,1, and 16. The header plus lines 2 and 15 would route to unmatched because of the leading "*" which does not match your regex. Result would be a FlowFile with: 0 abc-lr1-0 35189 20-Dec 03:43:54
1 abc-rr2-g 35209 20-Dec 03:43:54
16 abc-lr1-0 34631 20-Dec 03:43:54 Couple things to check if what you are seeing is entire original FlowFile getting routed to the "Original" and "Unmatched" relationships: 1. RouteText processor configuration. If i understand your use case correctly, it should be configured like this: 2. I noticed you sample data has leading and trailing whitespace so make sure processor is configured to ignore those. 3. Since you intent is produce a new FlowFile with only the lines matching the regex, make sure you set the above Routing Strategy. 4. Make sure the correct matching strategy is selected. Should be what I have above. 5. Click on the "+" to add a new dynamic property for your regex, The property name becomes a new relationship on the processor where your matching lines will be routed. 6. Since you are evaluating the source FlowFile content line-by-line, make sure your regex does not have a line return at the end of it. Correct: Incorrect (notice the line 2 which indicates a line return at end of regex): When I ran a little test flow using your sample data and regex, I got the desired results: The "lines" relationship has one new FlowFile with content of only the 3 matching lines The "unmatched" relationship contains a new FlowFile with content containing all the unmatched lines. The "original" relationship contains the original FlowFile that was processed by this processor. If you don't care about the original or unmatched FlowFiles, you can simply auto-terminate those relationships instead of routing them out of the processor in connections as I did above. Hope this helps, Matt
... View more
01-12-2021
05:49 AM
@Gcima009 It might be helpful to have more context around your issue. What action and/or NiFi component are you using when the exception occurs/ What NiFi version are you using? Can you share the entire error log from the nifi-app.log? Thanks, Matt
... View more
01-12-2021
05:34 AM
@Raj123 I am not a java developer, but NiFi is written in Java and the source code is open sourced. You would need to look at the code for the CSVReader to see how it handles AVRO schema inference. Sorry that I cannot be of more help in this specific query.
... View more
01-12-2021
05:21 AM
@CristoE Since this question already has an accepted solution and is specific to DISTCP replacement for HDFS, It would be much better to start an entirely new question in the community. You can always add a link to this question solution as reference in your new question post. You would get more visibility that way and we would not dilute the answer to this question with suggestions related to ADLS rather then HDFS.
... View more
01-08-2021
01:12 PM
@Raj123 NiFi offers many "record" based processors that support various record readers and writers. Those record readers have the ability of inferring an avro schema from the incoming record and the record writer can be configured to write the inferred schema to an attribute on the outgoing FlowFile. There is no specific infer schema processor for CSV source data. That would require a custom processor (perhaps one that utilizes the existing CSVReader controller service. Typically you would use a record based processor to manipulate, split, validate your record, so I am not the value or use case fro only wanting to infer the avro schema. That being said, you can get that inferred schema for example by simply using the "ConvertRecord" processor with a "CSVReader" (configured to infer schema) and a "CSVRecordSetWriter" (configured to "set avro.schema' attribute"). The written FlowFile will be same as source FlowFile but it will have an additional "avro.schema" attribute on the FlowFile containing the inferred avro schema. ConvertRecord: CSVReader: CSVRecordSetWriter: Hope this helps, Matt
... View more
01-05-2021
11:19 AM
@garoosy You should look in to using the " ExecuteSQLRecord" instead of "ExecuteSQL" for large volume data. To be efficient here you would have many records in a single FlowFile. Right now you have a single record per each FlowFile which is not going to be very efficient. The only way for "ExecuteSQL" to handle multiple FlowFile executions in a single connection is if the SQL statement used in every FlowFile is identical. In order to do that the unique values would need to come from FlowFile attributes. You may find these post helpful: https://community.cloudera.com/t5/Support-Questions/Nifi-ExectueSQL-how-to-force-a-parameter-to-be-a-string/td-p/240117 https://stackoverflow.com/questions/63330790/using-nifi-executesqlrecord-with-parameterized-sql-statements If you have threads that never seem to complete (will see small number in upper right corner of processor (2)), it is best to get a series of thread dumps (4 - 6) to verify thread is not progressing. Then you have to determine if what the thread is waiting on. Did you try setting a "Max Wait Time" on the processor? It defaults to 0 which means it would wait forever. Hope this helps, Matt
... View more
01-05-2021
10:49 AM
@kiranps11 Did you add and start a " DistributedMapCacheServer" controller service running on port 4557? The " DistributedMapCacheClientService" controller service only creates a client that is used to connect to a server you must also create. Keep in mind that the DistributedMapCacheServer does not offer High Availability (HA). Enabling this controller services will start a DistributedMapCacheServer on each node in your NiFi cluster, but each of those servers do not talk to each other. This is important to understand since you have configured your DMC Client to use localhost. This means that each node in your cluster would be using its own DMC server rather than a single DMC server. For a HA solution you should be using an external map cache via one of the other client offerings like "HBase_2_ClientMapCacheService " or "RedisDistributedMapCacheClientService", but this would require you to setup that external HBAs or Redis server with HA yourself. Hope this helps, Matt
... View more
01-05-2021
10:29 AM
1 Kudo
@Boenu Beginning with Apache NiFi 1.12 a change was implemented that changed the default anonymous access to static resources to false. This was done as part of https://issues.apache.org/jira/browse/NIFI-7170. This change also added the ability to add an additional property to the nifi.properties file to restore the behavior to that of Apache NiFi 1.11 and older versions: nifi.security.allow.anonymous.authentication=true The above restores previous behavior while work is done to change the how NiFi handles access to these static endpoints. The following Jiras cover that work: https://issues.apache.org/jira/browse/NIFI-7849 https://issues.apache.org/jira/browse/NIFI-7870 Hope this helps, Matt
... View more
01-04-2021
08:30 AM
@kalhan While it is possible to have a single ZK cluster to support multiple services, It is the recommendation that NiFi have its own dedicated ZK cluster. NiFi cluster stability is dependent on ZK and many of the NiFi processors that can be used depend on on Cluster state which is also stored in ZK. IF ZK becomes overburdened it can affect overall stability and performance of NiFi. If you found any of the answers provided on this query helped you, please select "accept solution" on each of them. Thank you, Matt Hope this helps.
... View more
01-04-2021
08:23 AM
1 Kudo
@adhishankarit As I mentioned you need an additional unique attribute that you only add on the failure path (Construct HDFSError UpdateAttribute) to MergeContent. overall-status = ERROR Since this attribute (overall-status) is not being set on the success path, the mergeContent "Attribute strategy" set to " Keep All Unique Attributes" will then set this overall-status attribute on the merged FlowFile produced. Keep All Unique Attribute --> any attribute on any FlowFile that gets bundled will be kept unless its value conflicts with the value from another FlowFile. Since you are not setting this attribute on your success path FlowFiles, it would only be set on mergeFlowFiles where one of more FlowFiles traversed the failure flow path. This allows you to capture overall-status of the zip bundle. Then in your ReplaceText processor you would use a more complex NiFi Expression Language (EL) in your replacement value. Something like: ${uniquefile}:${overall-status:isNull():ifElse('success','${overall-status}')}:${message} This will set "success" if the "overall-status" attribute does not exist on any FlowFiles that were part of the merged FlowFiles; otherwise it will set the it to the value set in the "overall-status" attribute. If you found this help, please take a moment to click "accept solution" on all responses that helped. Matt
... View more
12-28-2020
10:33 AM
@adhishankarit You dataflow screenshot does not reflect the entire dataflow you are then trying to describe making this use case hard to follow. 1. Your flow starts with a single zip file? 2. You unzip that file to produce numerous output FlowFiles? 3. You use load balanced connections to distribute all the produced FlowFiles across all nodes in your cluster? 4. Then you modify content of FlowFile using convertAttribitesToJson processor (Destination=flowfile-content)? Looks like you route the "success" relationship twice from this processor which means you have cloned your FlowFiles. Why? 5. One of these connection looks like it uses a Load Balance connection (how is it configured?) to feed a MergeContent. MergeContent can not merge across multiple nodes (can only merge FlowFiles on same node. How is MergeContent configured? Your desire output does not look like Json, but you are using AttributesToJson processor? 6. Where do the "failure" FlowFiles get introduced in to this dataflow? When you unpack your original FlowFile each produced FlowFile will have new attributes set on it to include segment.original.filename, fragment.identifier, fragment.count, and fragment.index. These attributes can be used with the "defragment" merge strategy in MergeContent. So I would avoid cloning FlowFiles post unpack. Process each FlowFile in-line. When you encounter a "failure", set an attribute on these FlowFiles only that states a failure occurred (successfully processed FlowFiles should not have this unique attribute). Then use MergeContent and set keep all Unique Attributes. This will allow the unique attribute if exists on any one FlowFile to show up on the output merged FlowFile will not work if same attribute exists on multiple FlowFiles with different values). Now after merge you can modify the content again using ReplaceText processor configured with Append to add the first line with overall status of this file from that unique attribute you preserved through the merge. Also not following statement: " also noticed that if there is a delay in processing " Hope this helps, Matt
... View more
12-28-2020
09:59 AM
@JelenaS You would need to share some screenshots of the policies/permissions you have set on the bucket(s) you have created in your NiFi-Registry. - Go to "Settings" (wrench icon upper right corner within NiFi-Registry - under "BUCKETS" click pencil icon for bucket you expect your user to see - Your NiFi user which is logged in to NiFi should have write,delete,read on the bucket. Would also be helpful to what "Special Privileges" you have set for each of your NiFi nodes inside NiFi-Registry as well. - Go to "Settings" (wrench icon upper right corner within NiFi-Registry - under "USERS" click pencil icon for each of your NiFi nodes - Each of your NiFi nodes (case sensitive) should have "Can proxy user requests" and read on "Can manage buckets" checked.
... View more
12-24-2020
07:35 AM
@adhishankarit There is nothing you can pull form the NiFi Rest-API that is going to tell you about successful outcomes form processor execution on a FlowFile. Depending on data volumes, this also sounds like a resource expensive endeavor. That being said, NiFi does have a Site-To-SIte (S2S) Bulletin reporting task. When a processor throws an error it will produce a bulletin and this reporting task can capture those bulletins and send them via NiFi's S2S protocol to another NiFi instance directly into a dataflow were you can handle via dataflow design however you like. Only way you can get INFO level logs in to bulletins is by setting the bulletin level to INFO on all you processors. This only works if you have also configured your NiFI logback.xml so that all NiFi components log at the INFO level as well. Downsides to this: 1. Every processor would display the red bulletin square in upper right corner of processor, This makes using this to find components that are having issues difficult 2. This results in a lot of INFO level logging to the nifi-app.log. You mention edge nodes. You could setup a tailFile processor that tails the nifi-app.log and then send via a dataflow that log data via FlowFiles to some monitor NiFi cluster where another dataflows parses those records via a partitionRecord processor by log_level and then routes based on that log_level for additional handling/notification processing. Downside here: 1. Since you what to track success, you still need INFO level logging enabled for all components. This means even this log collection flow is producing log output. So large logs and logs being written to even when actual data being processed in other flows is not happening. NiFi does have a master bulletin board which you could hit via the rest-api, but this does not get you past the massive logging you may be producing to monitor success. https://nifi.apache.org/docs/nifi-docs/rest-api/index.html Hope this gives you some ideas, Matt
... View more
12-24-2020
07:15 AM
@adhishankarit This only works if intent is to replace or append the FlowFile Attributes to the content of the FlowFile. But you will still not result in a single FlowFiles will all 15 attributes. ReplaceText does not merge anything and only has access to FlowFile Attributes on the FlowFile that is being executed upon. You would still need to merge the content of those 2 FlowFiles to have a single FlowFiles with all 15 attributes which would exist in the content of the new FlowFile.
... View more
12-24-2020
07:10 AM
@wasimakram054 First we need to answer some questions here: 1.What about the content of these two FlowFiles? Do both have same content? How do you want to handle that when you merge these FlowFIles? 2. Are all 15 of these attributes unique? Meaning that the same attribute name does not exist on both FlowFiles. Scenario 1: Let's assume both FlowFiles have same content. 1. Then you could user modifyBytes processor to remove all content from only one of those FlowFIles. 2. Then use Merge Content processor to merge those two FlowFiles and set property Keep all unique attributes. This will result in one FlowFile with content from only FlowFile that had content and all unique attributes from both source FlowFiles set on output new FlowFile. Scenario 2: Each FlowFile has unique content. 1. You could still use MergeContent just as in Scenario one and the resulting FlowFile will have all teh unique attributes and the merged content from both source FlowFiles. Another option: You could use maybe the putDistributedMapCache processor to write the desired attributes to a cache server. Then use the FetchDistributedMapCache to consume the needed attributes and place them on the other FlowFile. This will not perform as well and you need to consider under volume that you only process one set of FlowFiles at a time. Can be done but not as performant and adds complexity to dataflow design. Hope this helps, Matt
... View more
12-23-2020
01:39 PM
@Anurag007 You did not share how your logs are getting in to your NiFi. But once ingested, you could use a PartitionRecord processor using one of the following readers to handle parsing your log files: - GrokReader - SyslogReader - Syslog5424Reader You can then use your choice of Record Writers to output your individual split log outputs. You would then add one custom property that is used to group like log entries by the log_level This custom property will become a new FlowFile attribute on the output FlowFiles. You can then use a RouteOnAttribute processor to filter out only FlowFiles where the log_level is set to ERROR. Here is a simple flow I created that tails NiFi's app log and partitions logs by log_level and and then routes log entries for WARN or ERROR. I use the GrokReader with the following GrokExpression %{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:thread}\] %{DATA:class} %{GREEDYDATA:message} I then chose to use the JsonRecordSetWriter My dynamic i added property in the Property = log_level
Value = /level In my RouteOnAttribute processor, I can route based on that new "log_level" attribute that will exist on each partitioned FlowFile using two dynamic property which each become a new relationship: property = ERROR
value = ${log_level:equals('ERROR')}
property = WARN
value = ${log_level:equals('WARN')} Hope this helps, Matt
... View more
12-23-2020
12:48 PM
@JelenaS You are correct that this sounds like an authorization issue. I recommend tailing the nifi-registry-app.log and then perform the action of trying to version control a Process Group within NiFI's UI. How are you handling user authorization in your NiFi and NiFi-Registry? - File based authorization (users.xml and authorizations.xml) What identity.mapping patterns have you configured in your NiFi and NiFi-Registry? How are you authenticating your user that access both NiFi and NiFi-Registry? The only buckets that would be returned are those buckets for which the Authenticated user in NiFi has access to in NiFi-Registry. Keep in mind that the user/client strings in NiFi that are passed to NiFi-Registry must match exactly. Nodes will pass their full DN when they proxy the request on behalf of the authorized user. The user string will passed as is. That means identity mapping patterns will be applied on NiFi-Registry side against those NIFi DNs. Resulting mapped value must match the client string add as a user in NiFi-Registry. The passed user string must match exactly (case sensitive) or it is treated as a different user. Hope this helps, Matt
... View more
12-23-2020
08:28 AM
@pjagielski It is always helpful to share the exact NiFi version you are running as there may be known issues we can point you. Assuming here that you may be running latest Apache NiFi 1.12 release, my first thought may be related to this issue: https://issues.apache.org/jira/browse/NIFI-7992 While your content repo is not filling up, I would suggest inspecting you logs to see how often Content claims are being moved to archive. A background thread then removes those claims as a result of your archive settings. Hope this helps, Matt
... View more
