Hi @Rohit1997jio , Not sure if this can be done using the FlowFile Expiration. However, if you are using Nifi 1.16 or higher you can take advantage of another methodology using the  "retry" option on the target processor failure relationship as follows: The concept here is to use the settings for "Number of Retry Attempts" , " Retry Back Off Policy" & "Retry Max Back off Period" to configure how often and for how long the file is retried before it gets pushed to the failure relationship queue where you can then log the needed message. Every failed retry, the flowfile will be pushed back to the upstream queue and wait the designated time before its tried again. The challenge here is how to set those values so that the flowfile is only kept for certain period of time ( 1 hour in your case) specially the file will wait in the queue before its tried again depending wither you set the policy to Penalize or Yield, which is a good thing because you want to have some delay before the flowfile is tried again to avoid a lot of overhead. For example if you want the file to expire in an hour , and you want to try it 60 times where each time you wait 1 min before the next retry then you can set the values as follows: Number Of Retry Attempts:  60 Retry Back Off Policy : Penalize ( Set the Penalty Duration under Settings Tab to  1 min) Retry Maximum Back Off Period: 1 min ( this to ensure that the wait time in the queue doesnt exceed the initial penalty duration because every subsequent retry the duration penalty time is doubled - not sure why- ) In this case the flowfile will be retried 60 times upon failure , where each time the flowfile is pushed back to upstream queue an wait only max 1 min  before the next retry , which makes the total time flowfile is retried = 60  * 1 min  = 60 mins = 1 hour Depending how often you want to retry and how long you want to wait before the next retry, you can adjust those numbers accordingly. Once done with all the retries the flowfile will be moved to failure relationship where you can log the final message. If that helps please accept solution. Thanks     ... View more

Hi @glad1 , Can you elaborate more on the data that you want to remove? For example if the data is part of the CSV and it has unique value in one or more columns, then you can use QueryRecord processor where the query exclude records with this unique value. If the data is out of the CSV - like a header information - then depending how this data look like and if its surrounded with some special characters then you can use ReplaceText Processor with regex that would isolate those lines and then replace them with empty space and so on. If you can provide some sample data it would help in figuring out the best solution for this scenario. Thanks ... View more

Can you provide more details what the issue is  , what the error message you are getting if any and where and in which processor is causing the problem based on the provided input and expected output. I have never used PutDynamoDB but here are some links that can help: https://www.youtube.com/watch?si=ctBH-f-JOzAPgKAJ&embeds_referring_euri=https%3A%2F%2Fwww.google.com%2F&source_ve_path=MzY4NDIsMTM5MTE3LDEzOTExNywyODY2NCwxNjQ1MDY&feature=emb_share&v=Aw6PCz8gbmA   https://stackoverflow.com/questions/45840156/how-putdynamodb-works-in-nifi       ... View more

So Im not sure what are you going to use for inserting the Data into EleasticSearch/Slor. Its possible that you want to use concurrent curl api command with ExecuteStreamCommand processor. Alternatively, you can use  InvokeHttp processor to construct the API call which comes equipped with a lot of options and parameters to make the API call much easier. Nifi also comes with out of the box processor to interact with ElasticSearch & Solr like: PutElasticsearchHttp, PutElasticsearchJson, PutSolrRecord,PutSolrContentStream....etc. This can be much easier to setup than using API. If you are looking to do multiple upload concurrently, regardless of what process you use,   you can configure that on the processor itself by right click on the processor, select Configure , select  the  Scheduling tab and set the proper value in the "Concurrent Tasks" box, for example you can set 20: Keep in mind the following when you set this value : 1- Single Node vs. Cluster: if you are using single node then you can set the value of Concurrent Tasks as much as needed , but if you have a cluster and you are using some Load Balancing before making the upload to the target system , then you are already doing some concurrency where a given record might be uploaded at the same time by different nodes and you have to take that in consideration. For example, if you have 5 node cluster and 20 record per batch then each node will take 4 records and in this case the Concurrent Tasks will have the value of 4. 2- Make sure the "Maximum Timer Driven Thread Count"  is set to  >= 20 to allow Nifi to process 20 tasks concurrently , by default on this value is set to 10. To get to this value click on the 3 bar icon on top right , select Controller Settings and the value should be under the first tab "General": Recommendation on how to set this value: https://community.cloudera.com/t5/Support-Questions/How-to-configure-NiFi-to-maximize-the-usage-of-system-cores/td-p/189581   When it comes to batching the  data from the source database , you can use a processor like ExecuteSQL or ExecuteSQLRecord to fetch data out. in those processors configuration you need to setup the DB Connection Pool service to create the connection, for more information on this refer to : https://community.cloudera.com/t5/Support-Questions/how-to-configure-and-connect-mysql-with-nifi-and-perform/m-p/109478 https://www.youtube.com/watch?v=fuEosO24fgI Also in the configuration you can specify that you want 20 max records per flowfile in the "Max Rows Per Flow File" so that you dont get all 2 million in one file which might take a lot of time and memory to process which can result in an error depending on your heap size. ExcuteSQL will give you the result in Avro format. You  can use ConvertAvroToJson processor if you want to convert to Json, or use ExecuteSQLRecord processor and set the RecordWriter to the desired target format.   If you want to ensure that you are processing 20 records at a time and you want to prevent any records from being processed before the older batch is complete , you can use the Batch Processing at the Group Level as described here: https://www.youtube.com/watch?v=kvJx8vQnCNE&t=296s So the idea is you put processing & uploading the 20 records in a group where you configure the "Process Group FlowFile Concurrency" as described in the video above. If your ExecuteSQL fetch 20 rows per flowfile then you allow one flowfile at a time to enter the group. Inside the group you need to split the records and upload concurrently. Hopefully that will at least get you going. If you have any questions please let me know. If you find this helpful please accept solution. Thanks           ... View more

Hi @Sipping1n0s , Can you provide more information on the other system? How do you batch insert 20 records? is this an API call or some database? Thanks   ... View more