Member since
07-30-2019
3406
Posts
1622
Kudos Received
1008
Solutions
My Accepted Solutions
| Title | Views | Posted |
|---|---|---|
| 185 | 12-17-2025 05:55 AM | |
| 246 | 12-15-2025 01:29 PM | |
| 183 | 12-15-2025 06:50 AM | |
| 277 | 12-05-2025 08:25 AM | |
| 464 | 12-03-2025 10:21 AM |
02-11-2021
07:57 AM
@adhishankarit When moving on to a new issue, I recommend always starting a new query for better visibility. (for example, someone else in the community may have more experience with new issue then me). As far as your new query, your screenshots do not show any stats on the processor to get an idea of what we are talking about here in terms of performance. How many fragments are getting merged? How large are each of these fragments? NiFi nodes are only aware and have access to the FlowFiles on the individual node. So if node a is "out" (not sure what that means), any FlowFiles still on node "a" that are part of the same fragment will not yet be transferred to node b or c to get binned for merge. The Bin can not be merged until all fragments are present on the same node. Since you mention that bin eventually happens after 10 minutes, tells me that eventually all fragments eventually make it on to the same node. I suggest the first thing to address here is your space issue on your nodes. Also keep in mind that while you have noticed that node "a" has always been your elected primary node, there is no guarantee that will always be the case. A new Cluster Coordinator and Primary node can be elected by Zookeeper at anytime. If you shutdown or disconnect currently elected primary node "a" you should see another node get elected as primary node. Adding node "a" back in will not force ZK to elect it back as primary node. So don't build your flow around a dependency on any specific node being primary node all the time. Matt
... View more
02-10-2021
10:27 AM
1 Kudo
@bsivalingam83 The ability to "ignore" properties in various NiFi config files was added with the CFM 1.0.1 release. With older CFM versions (1.0.0) you can set a safety valve to overwrite the current set java.arg.13 value with something else: Above simply defines a key=value pair which would simply not be used by NiFi bootstrap. End result, NiFi no longer using G1GC and instead using the default Garbage Collector for your version of Java being used. Hope this helps, Matt
... View more
02-10-2021
09:55 AM
@has The listFile processor does not accept an inbound connection. If you know the filename of the file being created and the path where that file is created, all you need is the FetchFile processor. handleHTTPRequest ---> <flow to execute jar> --> updateAttribute (set path and filename attributes) --> FetchFile --> handleHttpRequest --> <rest of dataflow> The handleHttpRequest processor does not return content. It simply returns a response code back to for the original request which has not yet been responded to. The only thing you have control over in the response is the status code sent and sending custom headers in the response. Hope this helps, Matt
... View more
02-10-2021
09:03 AM
3 Kudos
@Jarinek The process really depends on what update you are trying to make. 1. You can not remove a connection that has queued FlowFiles in it, but you can redirect it to a different target processor with queued data. 2. You can not redirect a connection if the processor it is currently attached to still has a running thread. Stopping a processor does not kill threads, it simply tells the processor to not execute again at the configured run schedule. Existing threads will continue to run until they complete. Until all threads exit, the processor is still in a state of "stopping" even though UI reflect red square for "stopped". 3. You cannot modify a processor if is still has running threads (see note about "stopping" processors above) 4. If you stop the component that is on the receiving side of a connection, any FlowFiles queued on that connection, not tied to any active thread still running on target processor component, will not be processed and remain queued on the connection. You can manual empty a queue through a rest-api call (means data loss), but that is not necessary if you are not deleting the connection. Attempts to perform configuration changes when components still have active threads or are in a running state will result in an exception being thrown and the change not happening. Attempts remove connections that have queued FlowFiles will throw an exception and block removal. Now if all you are trying to do is modify some configuration on a processor, all you need to do is stop the processor, check that it has no active threads, make the config change, and then start the processor again. Not sure wha you are asking with "update the flow ignoring any data in failure or error connection queues". NiFi does not ignore queued FlowFiles. It also not wise to leave connection with queued FlowFiles just sitting around your dataflows. Those old queued FlowFile will prevent removal or content claims that contain that FlowFiles data. Since a content claim can contain the data from 1 to many FlowFiles, this can result in your content repository filling up. NiFi can only remove content claims which have no FlowFiles pointing to them anymore. Here are some useful links: https://nipyapi.readthedocs.io/en/latest/nipyapi-docs/nipyapi.html https://github.com/Chaffelson/nipyapi http://nifi.apache.org/docs/nifi-docs/rest-api/index.html https://community.cloudera.com/t5/Community-Articles/Update-NiFi-Connection-Destination-via-REST-API/ta-p/244211 https://community.cloudera.com/t5/Community-Articles/Change-NiFi-Flow-Using-Rest-API-Part-1/ta-p/244631 Hope this helps, Matt
... View more
02-09-2021
05:52 AM
@medloh That is the correct solution here, the filename is always stored in a FlowFile attribute named "filename". Using the updateAttribute processor is the easiest way to manipulate the FlowFile attribute. You can use other attributes, static text, and even subjectless functions like "now()" or "nextInt()" to create dynamic filenames for each FlowFile. https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html Hope this helps, Matt
... View more
02-09-2021
05:48 AM
@Umakanth The GetSFTP processor actually creates a verbose listing of all Files form target SFTP for which it will be getting. It then fetches all those files. Unlike the ListSFTP processor, the getSFTP is an older deprecated processor that does not store state. My guess here is that at times the listing is larger then other times or as you mentioned some occasional latency occurs resulting in enough time between creating that list and actually consuming the files, that the source system has moved the listed file before it is grabbed. In that case moving to the newer ListSFTP and FetchSFTP processors will help in handling that scenario. The listing will list all the files it sees and the FetchSFTP will fetch the content for those that have not yet been moved by the source system. The FetchSFTP will still throw an exception for each file it can not find still and route those to the not.found relationship which you can handle programmatically in your NiFi dataflow(s). Thanks, Matt
... View more
02-08-2021
08:45 AM
@medloh The Schema only needs to be defined in the RecordReader configured in the PutParquet processor. In the case of the ConvertRecord processor there exists both a Record reader and a Record Writer. You can have the RecordReader get the Schema from the Record Writer or define its own Schema. Hope this helps, Matt
... View more
02-08-2021
08:20 AM
@Jarinek NiFi Variables can only be used by component properties that support NiFi's Expression Language (EL). NiFI Parameters can be used in ANY component property including those that are encrypted. This gives more flexility to users, especially those users who use NiFi-Registry to promote version controlled Process groups across multiple NiFi instances/clusters. It is often the case that different environments have different URLs and passwords in use within the same dataflows. A Dataflwo can thus be promoted to another environment that simply uses different Parameter values thus not requiring the user to update a large number of components each time a new version of flow is promoted from one environment to another. You are correct that Parameters are similar to Variables in respect to assignment to a process group. You can only have one Parameter context assign to a Process Group. Hope this helps, Matt
... View more
02-05-2021
08:09 AM
1 Kudo
@medloh The article you are using for reference is old and bit out of date. As part of the work that went into NIFI-3921 the schema properties within the putParquet processor were removed. Before these changes were made at the time of that article you referenced, you had to set the schema properties and they had to match the schema properties set in the recordReader. With the changes the processor simply gets them from the reader so they do not need to be configured a second time in the processor properties. Also at time of that article there was no parquetReader or ParquetRecordSetWriter controller services. Now that NiFi has a ParquetReader and writer, you can use the ConvertRecord processor to read a source FlowFiles and convert it parquet within your dataflow and then have freedom to use whatever processor you want downstream in your dataflow to write out the parquet content. You can think of the putParquet as a combination of ParquetRecordSetWriter and putHDFS with a selectable recordReader only. Hope this helps, Matt
... View more
02-05-2021
06:25 AM
1 Kudo
@Kamil_Valko All NiFi logging is handled by the logback.xml file. By default you will find that NiFi's logback.xml contains three unique file appenders. 1. APP_FILE 2. USER_FILE 3. BOOTSTRAP_FILE. To log anything to a different file, the first thing to do is create another appender. I recommend copying the APP_FILE appender. <appender name="FTP_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/nifi/nifi-ftp.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/nifi/nifi-ftp_%d{yyyy-MM-dd_HH}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>5</maxHistory>
<totalSizeCap>1GB</totalSizeCap>
</rollingPolicy>
<immediateFlush>true</immediateFlush>
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
</encoder>
</appender> Then you need to add a new logger for the FetchFTP processor class. You cannot control logging for specific instance of a processor. This means all logging from all instantiations of the listFTP processor on the canvas will get handled by this new logger. <logger name="org.apache.nifi.processors.standard.FetchFTP" level="ERROR" additivity="false">
<appender-ref ref="FTP_FILE"/>
</logger> You can turn off logging for a specific processor class, but I agree with @ckumar that this is not an ideal path. But for completeness of this response, setting level to "OFF" on a logger for a specific class will stop all logging from that class. (Remember that this affects logging from all occurrences of this processor used by this NiFi) Hope this helps, Matt
... View more