Hi, You can use the ReplaceText Processor as follows: If you find this helpful please accept solution. Thanks ... View more

Hi ,   Is not every record you pull through the GenerateTableFetch will have the max value for the max column? ... View more

Hi, The full picture is not clear on how your flow works, for example where do you get the json that goes through the jolt processor and input port from? Is there a correlation between getting the json from the different sources where one json is generated before the other or the two json sources are independent of each other? If they are independent of each other then how do you know when to merge , for example what if the json from the input port did not come? Can you explain more on how your flow works and if there is a dependency where one json input drives the other or not? ... View more

Hi, Couple of options I can think of but that depends on your flow and what are you trying to do with the data:   1- Get all the parameters you need through the whole flow from the xml after you fetch the file. Those parameters will be there on every generated flow file later on and at any step you can retrieve them and do whatever needed. 2- If the data you are trying to get downstream from the xml is unrelated you can always have multiple branches after the FetchFile and created different flow for each branch that can run asynchronously. 3- IF you are looking to do different API calls based on the same XML, the InvokeHttp has an Original relationship besides the Response relationship in case each response independent of the others. 4- If you are using Nifi 1.16 and higher and you are looking for data enrichment for the base XML you can look into the processors : ForkEnrichment, JoinEnrichment.     ... View more

Hi, In the ExecuteSQL or ExecuteSQLRecord you have Pre Query, Select Query and Post Query properties. In the Pre Query can you declare variables to store the output parameters, run the stored proc passing those parameter and then insert them into a temp table which then you can use the Select Query property to read the values from the temp table and that will translate to flowfiles. If you are using the ExecuteSQLRecord you can define in which format you want the flowfile in by defining the record writer property. Hope that helps. ... View more

Hi, not sure if you are repeating the attribute name in the processor dynamic property values! Make sure the different header attributes are set\add as follows:     ... View more

Hi, you can use  ReplaceText by setting the Evaluation Mode property to "Line-by-Line", then set the "Line-by-Line Evalaution Mode" to "Last Line". In the Replacement  Value set it to Empty String Set and the Search Value leave as "(?s)(^.*$)" as seen in the screenshot below:   If you find this helpful please accept solution. Thanks   ... View more

The lookupRecrod processor allows you to specify multiple lookup columns in the DatabaseRecordLookupService incase the id is not enough. I had a similar sitution and I usually defer this kind of check to the DB, so my workflow will call stored proc - with the record parameters - that will check if the record exist or not and based on that decide wither to update or insert. This way Im making only one connection to the DB. ... View more

Hi ,   What is the flowfile content format? Is there an Id you can check using the LookupRecord processor to see if the record already exists or not? ... View more

Can you explain how did you resolve it with XMLRecrodSetWriter? Thanks ... View more

Hi, Im not sure this can be fixed by changing some configuration on what you have wither on the processor or the service level. Since the root element is not considered in the conversion from xml to Json, as workaround you can surround your xml with arbitrary root element with the ReplaceText Processor before the ConvertRecrod Processor:   Search Value: (?s)(^.*$)   Replacement Value: <root>$1</root>   Evaluation Mode: Entire Text  This will output the following xml: <root>n<a> <b>45</b> </a> n</root> And After ConvertRecord Prceossor : {"a": {"b":45}} Hope that helps. If it does, please Accept Solution. Thanks               ... View more

Hi the only way I can think of to avoid duplicating the content of the flowfile and not having to worry about the Maximum Buffer Size in the ExtractText processor is to use the ExecuteScript processor to update the flowfile content into json format. Here is an example of Script using "ECMAScript" in the ExecuteScript Processor "Script Body" property that can do that: var StreamCallback = Java.type("org.apache.nifi.processor.io.StreamCallback"); var IOUtils = Java.type("org.apache.commons.io.IOUtils"); var StandardCharsets = Java.type("java.nio.charset.StandardCharsets"); var flowFile = session.get(); if(flowFile != null) { // Create a new StreamCallback, passing in a function to define the interface method flowFile = session.write(flowFile, new StreamCallback(function(inputStream, outputStream) { var text = IOUtils.toString(inputStream, StandardCharsets.UTF_8) var obj = new Object(); obj.SomeKey = text; var jsonString= JSON.stringify(obj); outputStream.write(jsonString.getBytes(StandardCharsets.UTF_8)) })); } session.transfer(flowFile, REL_SUCCESS) Hope that helps. If it does please Accept Solution. Thanks ... View more

In this case I would suggest ReplaceText processor with the Search Value set to: (MapRecord|\[|\]) And Replacement Value is set to Empty String.This will produce the result: aaaa|"{country=CHINA, city=null, street=null, latitude=null, postalCode=null, geocodeAccuracy=null, state=null, longitude=null}" ... View more

Not sure I understand the question, which processor/property are you taking about? Basically if you want to access a flowfile attribute value in expression language you can use ${A} in case you have an attribute called A ... View more

Hi,   You can use FlattenJson processor to flatten your json before converting to CSV. The output Json after flattening will be something like this: { "id": "aaaa", "billingaddress.city": null, "billingaddress.country": "CHINA", "billingaddress.geocodeAccuracy": null, "billingaddress.latitude": null, "billingaddress.longitude": null, "billingaddress.postalCode": null, "billingaddress.state": null, "billingaddress.street": null }   Another Option is to use JoltTransformJSON to flatten your json if you don't like the "billingaddress." prefix. You can use the following Jolt Spec: [  {    "operation": "shift",    "spec": {             "id": "id",             "billingaddress": {             "*": "&"                 }       }   } ]   Which will produce the following json: { "id" : "aaaa", "city" : null, "country" : "CHINA", "geocodeAccuracy" : null, "latitude" : null, "longitude" : null, "postalCode" : null, "state" : null, "street" : null } If you think this solves the issue please accept solution. Thanks ... View more

Hi, If I understood you correctly you want to log some information when the QueryRecord is executed, in this case you can use the Original relationship coming out of the QueryRecord where you can log the information like filename,timestamp...etc. ... View more

Marfill, If backpressure is applied when the total number of flowfiles in a given queue has reached (#nodes * the limit per node) for example if you have a cluster of 3 nodes and the threshold is set to 10,000 then the backpressure will be applied when total # of flow files = 30,000 and so on. Regarding the Control Rate I believe its done per node statistics, for example if you have a control rate that allows 1 flow file per hour and the control rate processor is set part of load balancing on 3 nodes cluster, let say you receive total of 3 files for the first time  one on each node then the 3 will be get processed immediately. ... View more

Hi, Can you post the Expression Language you used? I'm seeing in the error message that its complaining about the "literal" function doesnt exist. Did you use it with the variable name that contains the original date value? If so then remove the literal function and just keep the variable since literal works with constants. So the expression should be something like this:   ${date_variable:replaceAll('(IST)\s',''):toDate('EEE MMM dd HH:mm:ss yyyy'):format('yyyy-MM-dd HH:mm:ss')} ... View more

Hi, If your configurations are similar to mine then I would suggest that you test it on version 1.16.0 to see if its a bug with version  1.16.2 which in this case needs to be reported. Another option for you is to write custom code using ExecuteScript processor that will generate a different flowfile  for each json record since the ExecuteProcess wont have this option and would write everything in the output steam into one flowfile. you can refer to the following post to help you generate multiple flowfiles: https://community.cloudera.com/t5/Support-Questions/Split-one-Nifi-flow-file-into-Multiple-flow-file-based-on/m-p/203387 Another option you can try incase there is a problem with the SplitContent processor is to create every json record in a different line and then try to split them using SplitText processor and set the property "Split Line Count" to 1 ... View more

Hi, I ran into this issue before when dealing with ISO timeformat and it would not allow me to write to the DB. In this case you have to use an updateAttribute to convert this format to the format that DB can accept (yyyy-MM-dd HH:mm:ss), this worked for me: ${literal('Wed Oct 13 15:58:58 IST 2021'):replaceAll('(IST)\s',''):toDate('EEE MMM dd HH:mm:ss yyyy'):format('yyyy-MM-dd HH:mm:ss')} Hope it would help. ... View more

Hi, I'm not aware of any out of the box processor that can help you with that. A suggestion would be to write custom code in an ExecuteScript Processor that gets the filename attribute and stores in a file with the required format in some staging directory, In this processor  you can also decide how many flowfiles you want to process per file and once that file reaches the limit (could be by file size or number of entries) you then move the file to final directory where and GetFile processor is reading from and direct the content of the read file (filenames) to PUTEmail. New file will be created in the staging area to address any new entries after moving the older file to the final directory. ... View more