Member since
07-29-2020
526
Posts
261
Kudos Received
157
Solutions
My Accepted Solutions
Title | Views | Posted |
---|---|---|
98 | 09-03-2024 12:36 AM | |
132 | 08-30-2024 06:23 AM | |
260 | 08-26-2024 04:39 PM | |
199 | 08-21-2024 12:01 PM | |
201 | 08-10-2024 03:18 AM |
09-07-2024
07:58 AM
Hi @xtd , Welcome to the community. you definitely look like newbie :), but dont worry, as this is common when you start with Nifi but with time and experience you get better for sure. There are many tutorials out there to help you get started that I would recommend going through to understand how nifi works and what are the best practices. However one crucial concept that every beginner needs to be familiar with early on can be found in series of youtube videos by Mark Payne who is one the creators of Nifi around Nifi Anti Pattern that I recommend you to watch. I can see many issues with this flow. Above all its not doing it the nifi way. Here are some notes on some of the main issues: 1- When you work with formats like CSV, Json , xml ...etc. You need to take advantage of Nifi out of the box capabilities to parse the data and not doing that using replace text which could be inefficient. 2- Its not recommended to use ExtractText to extract the whole content of a flowfile into an attribute when you cant control how big the content can be. you will hear that a lot that attribute are stored in JVM heap which could be limited and expensive which can cause problem if you have large content. 3-When passing parameters to your SQL query through file attribute the N placeholder in sql.args.N.value stands for the parameter index in the query , for example if you the following insert statement: insert into Table1 (col1, col2) values (?,?) Then the processor expects two set of sql attributes where for the first arg (N=0 ) and the second (N=1) as in sq1.args.0.value, sql.args.1.value...etc. The way I would design this flow depends on really what are you trying to do after the ExecuteSQL. The result of the ExecuteSQL if you pass in the list of ID's correctly and based on the the configuration will return a single flow file with the list of records matching the ID's in an Avro format, so what are you going to do with this result? My guess and based on similar scenarios you probably want to process each records on its own wither you want to do data enrichment or some API call ...etc. Based on that here is how I would design this flow: 1- Get the File from the source (GetFile, GetSFTP...etc.) 2- Use SplitRecord to to get every json record into singleflow and convert it into into format that is easy to parse and extract attributes using methods other than Regex in ExtractText, here is an example: Im converting to Json because I can extract the ID using json path as we will see in the next step. You can use the default setting for the reader and writer services. 3- User EvaluateJsonPath to extract the ID into an attribute. This way we know that the value we are going to extract is single integer value and we wont be concerned much about heap storing large content. 4- Fetch the record corresponding to each ID from SQL. I know you probably think that this is going to do sql transaction for each ID vs one select for all and maybe this is the pitfall of this design but those transaction are going to be short and simple and probably perform better specially when you have a cluster where do load balancing or you can increase the number of threads on the processor itself . As I mentioned above the ExecuteSQL will give you the result in Avro format which you cant do much parsing with unless you convert to other parsable format (json, csv, xml...etc) using ConvertRecord , however you can save the the need for ConvertRecord by using ExecuteSQLRecord instead of ExecuteSQL where you can set the record writer to whatever format before sending the result to the success relationship. Lets assume I want to convert the result back to CSV, the configuration would be: Once you get the record in the intended format you can do whatever needed afterward. I think you have enough to get going. Let me know if you have more questions. If you find this is helpful please accept the solution. Thanks
... View more
09-07-2024
12:59 AM
2 Kudos
Hi @shiva239 , I dont think there is a problem using the DuplicateFlowFile processor in production environment. If its really intended for test environment then it wont be provided as an option to begin with. Actually when I search the community for question around flowfile replication I found this post that mentions this processor as part of the solution but I dont see any comments advising against it siting test vs prod regardless if it helped in that case or not. However, if you are not comfortable using the Duplicate processor, there are of course different ways of doing the same thing in Nifi. Lets assume you want to write code to address this problem instead of using Nifi, how would you do it without having to repeat the same code for each target DB? I think one of the first things that comes to mind is using a loop. You can create a loop in Nifi in multiple ways, the easiest I can think of is to use the RetryFlowFile processor. Although its intended more for error handling you can still use it to replicate a loop flow. All you have to do after getting the file is to set the maximum number of retries to how many times ( or steps ) you want to execute against and then redirect the retry relationship to itself and the next step (assign DB & Table based on index). Once the number of retries exceeded you can handle that via the retries_exceeded relationship or you can simply terminate the relationship so its completed. The RetryFlowFile will set a counter on each retried flowfile which you can use to assign the DB and Table accordingly. Here is an example of simple flow that loops 5 times against a given flowfile input from GenerateFlowFile: This how the RetryFlowFile is configured: The Retry Attribute will store the retries count in the provided attribute name "flowfile.retries". If you run the GenerateFlowFile once and look at the LogMessage data provenance you will see it was executed 5 times against the same content : If you check the attributes on each event flow file you will see the flowfiles.retries is populated with the nth time the flowfile was retried. Keep in mind the data provenance stores the last event at the top which means the first event flowfile attribute will have the value of 5: hope that helps. If it does, please accept the solution. Thanks
... View more
09-03-2024
12:36 AM
3 Kudos
I think you are confusing the ExtractText and the ReplaceText proessors. The ExtractText doesn't have Search Value & Replacement Value properties but the ReplaceText does. That is why I said post screenshot would be helpful because have I known that its replace Text my answer would have been different. To get the desired result in this case , you need to specify the following pattern in the Search Value Property: ^(.{5})(.{10}).* Basically you need to specify the full lline text that you want to replace with the matched group. When you stopped at "^(.{5})(.{10})" it meant that you only want to replace up to the 15th character of the full text with the result $1,$2 and that is why you were getting the reminder of the text. By adding ".*" at the end it will replace the whole line and not just up to the 15th character. The final config will look like this I hope that makes sense.
... View more
09-03-2024
12:09 AM
1 Kudo
Hi,
My Apologies. I think I forgot to mention that in both cases you need to set the Timestamp Format in the CSVRecordWriter to the target format as follows since by default it converts the datetime to epoch time:
The point from the conversion in the QueryRecord is to tell the CSVReader that this is a datetime , however without setting the format in the writer it was converting it back to epoch time as the documentation states:
Setting the format there is critical to get the desired output.
Hope that helps.
... View more
09-02-2024
04:53 PM
2 Kudos
Hi, It could have been helpful if you were able to provide some examples regarding the different scenarios with what is expected vs what are you getting. Also providing screenshot of the processor\s in question can help making sure that you have the correct configuration to handle your case. One thing confusing to me is you dont mention anything about white spaces and if they count as a character in case of the name or the address or not. Going with what you provided, if we assume we have the following line: smithaddress123AAAA where name expected to be: smith (1-5) address: address123 (6-15) I have configured the ExtractAddress processor as follows (basically adding new dynamic properties to define the extracted attributes): The output flowfile will have the following attribute which what is expected: The reason on why you are getting additional attributes with an index is because how the processor works in breaking up matching group. You can read more about this here. If you find this helpful please accept the solution. Thanks
... View more
08-31-2024
10:19 AM
1 Kudo
Hi @NagendraKumar , Im not sure that you can use the function "DATE_FROM_UNIX_DATE" since according to the sqlcalcite documentation its not a standard function. If I may recommend two approaches to solve this problem: 1- Using Sql Calcite function TIMESTAMPADD: select TIMESTAMPADD(SECOND, 1724851471,cast('1970-01-01 00:00:00' as timestamp)) mytimestamp from flowfile 2- Using Expression Language: select '${literal('1724851471'):multiply(1000):format('yyyy-MM-dd HH:mm:ss')}' mytimestamp from flowfile In both cases you have to be aware of the timezone that the timestamp is converted into I think one uses local while the other uses GMT Hope that helps. If it helps please accept the solution. Thanks
... View more
08-30-2024
06:23 AM
1 Kudo
Hi @xtd , I did not try this but I think what is happening is that when you call the PutDatabaseRecord using CSVReader it doesnt differentiate null and empty string value and it will assume the later is also null and that is why you get the error. Have you tried converting csv into json and then do update record to see if that makes any difference? Another option is to use PutSql so that you can write your own sql but this would be cumbersome in case you have many fields.
... View more
08-30-2024
05:54 AM
1 Kudo
Hi @moshell281 , @steven-matisonis right, when you use infer schema the processor will map the value as it thinks the data type should be, so if it sees an integer it will map to an integer and so on. To force certain datatypes you have to define your own Avro schema and feed it to the Json Record Set Writer since this is your target format. If you are not comfortable working with Avro schema then you can use UpdateRecord or even JsonJoltTransformation processors to cast the value into the proper type. Hope you find this helpful.
... View more
08-27-2024
05:28 AM
Hi, Its not clear from your description what is the input json format and what is the expected output format, but from what I was able to understand from your description you have the following json: {
"sessionStopTime": "09/07/2024 06:47:19",
"resultCode": [
"2001",
"2001"
],
"recordOpeningTime": "09/07/2024 06:44:22"
} And you want to produce new attribute for the second array element of resutltCode to be stored in another field like this even though in your case both array elements are the same: {
"sessionStopTime": "09/07/2024 06:47:19",
"resultCode": [
"2001",
"2001"
],
"resultcodemscc":"2001",
"recordOpeningTime": "09/07/2024 06:44:22"
} If this the case then Jolt spec would look like this: [
{
"operation": "modify-overwrite-beta",
"spec": {
"resultcodemscc": "=elementAt(@(1,resultCode),1)"
}
}
] If that helps please accept solution. Thanks
... View more
08-26-2024
04:39 PM
2 Kudos
Hi @NagendraKumar , I think you misunderstood how the Stateful UpdateAtrribute works which is OK since a lot of people would think this way too if you have not used before which happened to me as well :). Basically when you make an UpdateAttribute Stateful by setting the Store State Property, as the value says it will "Store state locally" meaning you only can access the previous state of a given attribute within the processor itself. As I can see, you are trying to store the Batch Id in the Second Update Attribute and then access it from the first UpdateAttribute and that is why you are getting empty string because it doesnt exist there. You dont need two UpdateAttributes to manage that and one should do the job. Lets assume we have the following flow which consist of the following processors: 1- GenerateFlowFile: This will simulate setting new BatchId attribute by adding dynamic property BatchID and set it to some value. 2- UpdateAttribute: This will be a stateful and it will have two attributes: One to get the last saved BatchId value and another to set the last saved batch ID to the Current. 3- RouteOnAttribute: This is basically where you compare previous to current and route accordingly. Here is how the flow looks like Here is the config for each processor: GenerateFlowFile: UpdateAttribute: RouteOnAttribute: Basically, if you run once for the first time you will get the flowfile routed to the unmatched relationship of the RouteOnAttribute (since no previous value was set), however if you run it again without changing anything the result will be routed to the Match relationship since the previous saved value will equal the new one. Change the value in the GenerateFlowFile and it will go to unmatched and so on. I know you probably wondering how this works since in the UpdateAttribute Im referencing the LastSavedStateBatchID while at the same time its being set to the CurrentBatchID?! which comes first ? well the answer is simple: If you refer to the stateful documentation where its talking about you will find the following line: " If stateful properties reference other stateful properties then the value for the other stateful properties will be an iteration behind" . Which means PreviousBatchID will be set to the lastSavedStateBatchID before resetting the later to the current . if that makes any sense 🙂 There is a cleaner way of doing it which help eliminate this confusion and the circular reference by defining Rules under the Advanced feature which you can play with but I feel this way is much shorter. Hope that helps, if it does please accept solution. Thanks
... View more