@Anderosn  In-between your SplitJson and PuSQL processors are you rebalancing the FlowFile across multiple nodes in a NiFi cluster?  Are you routing any of the split Json messges down a different dataflow path that does not lead to this pusSQL processor? The reason I ask is because the splitJson processor will write the following FlowFile attributes to each new FlowFile created (each split): The fragment.identifier value and fragment.count are used by the putSQL processor when "Support FragmentTransactions" is set to "true" (default).  This means that, if not all split jsons are present at this putSQL and located on the same node of the NiFi cluster, the FlowFiles part of the same fragment.identifier will not be processed and remain on the inbound connection to the PutSQL. I'd start my listing the connection and checking these attributes to verify the fragment.count is "10", the fragment.identifier has same value on all 10, and fragment.index value shows numbers 1 to 10 across those 10 FlowFiles. If making sure all fragments are processed in same transaction is not a requirement for your dataflow, try changing "Support Fragmented Transactions" to false and see if these 10 FlowFiles get successfully executed by your putSQL processor.   If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt   ... View more

@kellerj  The alternative is making rest-api calls or navigating to every processor that writes state and purge it manually.  You would need to keep track of all your current and newly added processors that store state in order to accomplish, but shutting down your NiFi is not needed to clear state this way.  🙂  As far as finding all processors that store state, that is challenge in itself.  The embedded documentation for every processor will have a "State Management:" section which will tell you if the component stores state (Note that processors that show only "Cluster" state will store that state locally if it is a non cluster configured standalone instance of NiFi"). Once you have identified all the components your operations team is using by their unique UUID and filtered only those that write state, you can use that info to clear state using NiFi's rest-api, which is a multi-request process.  Then you need to worry about your operations team adding additional state based components later or removing and re-adding an existing state based processor triggering a new UUID. So while possible, it is challenging to orchestrate and maintain. You can also right click on a component to view its state and then purge that state directly from the listed state within NiFi UI. If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more

@JohnSilver  Your are not using the processor correctly resulting in your issue. The "list<type>" processors are designed to optimize distributed processing of files from sources that may not be NiFi cluster friendly.  They are designed to simply list the contents of the target input directory and produce a single 0 byte FlowFile with metadata about the content. Example of what FlowFile Attributes are created by the ListSMB processor: The list processors are intended to be configured to execute on "primary node" only in their configuration. This prevents all nodes in a NiFi cluster from listing the same files.   These 0 Byte FlowFiles can the be distributed/load-balanced across all the other nodes in the cluster using the load balanced configuration capability on a NiFi connection. Finally these now distributed 0 byte FlowFiles are sent to the Fetch<type> processor that should be configured to use the metadata/attributes on the FlowFile to retrieve the content for each listed FlowFile and insert in to the FlowFile content (after Fetch<type> processor, FlowFile size will no longer be 0 bytes).   Where you have misconfigurations: 1. You configured listSMB with "no tracking" which means it will retain no state on what was previously listed.  Without state, every execution of the listSMB will result in listing the same source files over and over again. 2. Each listed file becomes its own 0 byte FlowFile with a bunch of added attributes.  You are then passing that to your FetchSMB processor.  The default configuration of the "Remote File" processor property is "${path}/${filename}" which would take the value from the FlowFile attributes "path" and "filename" to fetch the content for the FlowFile.  You have instead misconfigured this property to always fetch the same content no matter which FlowFile is being processed.  So you are inserting the same content in to every one of your unique FlowFiles (each listed FlowFile has a filename that is persisted).  That is why you see same weight/size for all your fetched FlowFiles. You might try configuring your listSMB processor Input Directory to point at the full path to the single file you want to list.  I have not tried this as the intended usage is listing everything from within a target directory.  If that does not work, you could use a routeOnAttribute processor to route only the FlowFile with the specific filename you are looking for to FetchSMB and terminate unmatched. Also keep in mind the your ListSMB is by default going to re-list the same File(s) over and over because you have it configured with "no tracking" and default run schedule is "0 sec" (which means schedule to execute as fast/often as possible) Your PutHDFS appears to be working as expected, your issues seems purely in your upstream configurations.   If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more

@kellerj  Depending on your setup, state may be stored locally, in Zookeeper, or a mix of both. With a standalone non NiFi cluster setup, all state will be store in the local NiFi state directory. With a NiFi cluster setup (even if cluster only consists of one node), some processors will store state in a mix of local via file (state specific to a single node) and cluster via zookeeper (state that needs to be shared amongst all nodes in a NiFi cluster). The NiFi configuration file "state-management.xml" defines where both the local state and cluster state is being stored.  You can clear all local state by simply emptying the contents of the locally configured state directory on every node.  The "Directory" is defined in the "local-provider" within the state-management.xml file You can clear the Cluster state, by clearing out the znode on zookeeper defined in the "cluster-provider" within the state-management.xml file. NiFi must be shutdown before performing clearing out either local or cluster state.   If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more

@sam_s0ni  Apache NiFi never intended with the context of the NiFi framework to allow parameters to be dynamically accessed within NiFi Expression Language (NEL).  The EvaluateElString() exposed a bug in which this was possible while never intended.  That bug was addressed in apache NiFi 1.19 thus resulting in what you are seeing.   As an alternative, you could store your schemas in an external database and pull those schemas back dynamically based on a FlowFile attribute instead of using Parameter contexts to store those schemas. If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more

@MWM  MergeContent is probably not the processor you want to use to merge your JSON files as it will simply concatenate them together instead of creating one larger json.  You should use the MergeRecord processor instead.  In this processor you can specify a JsonTreeReader to read your source Json files and then use a CSVRecordSetWriter to output merged CSV file(s).  The rest of the MergeRecord processor configuration will control how many source Json records are added in to a single output CSV.   If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more