About ishantiwari91

ishantiwari91 · ‎04-28-2019

@Shu But in that case do I need to hard code the column names in EvaluateJsonPath?Kafka message and attributes are dynamic. Original message has more than 200 attributes. Is there any way where message can be filtered on the basis of structure and not on attribute names?

ishantiwari91 · ‎04-27-2019

I am consuming a nested Json Kafka message and after some transformations, storing it into hive. Requirement is json contains several nested arrays and we have to denormalize it so that each element in array forms a separate row in hive table. Would JoltTransform or FattenJson work or do I need to write groovy script for the same? Sample input - {"TxnMessage": {"HeaderData": {"EventCatg": "F"},"PostInrlTxn": {"Key": {"Acnt": "1234567890","Date": "20181018"},"Id": "3456","AdDa": {"Area": [{"HgmId": "","HntAm": 0},{"HgmId": "","HntAm": 0}]},"escTx": "Reload","seTb": {"seEnt": [{"seId": "CAKE","rCd": 678},{"seId": "","rCd": 0}]},"Bal": 6766}}} Expected output - {"TxnMessage.PostInrlTxn.AdDa.Area.HgmId":"","TxnMessage.PostInrlTxn.AdDa.Area.HntAm":0,"TxnMessage.HeaderData.EventCatg":"F","TxnMessage.PostInrlTxn.Key.Acnt":"1234567890","TxnMessage.PostInrlTxn.Key.Date":"20181018","TxnMessage.PostInrlTxn.Id":"3456","TxnMessage.PostInrlTxn.escTx":"Reload","TxnMessage.PostInrlTxn.Bal":6766} {"TxnMessage.PostInrlTxn.AdDa.Area.HgmId":"","TxnMessage.PostInrlTxn.AdDa.Area.HntAm":0,"TxnMessage.HeaderData.EventCatg":"F","TxnMessage.PostInrlTxn.Key.Acnt":"1234567890","TxnMessage.PostInrlTxn.Key.Date":"20181018","TxnMessage.PostInrlTxn.Id":"3456","TxnMessage.PostInrlTxn.escTx":"Reload","TxnMessage.PostInrlTxn.Bal":6766} {"TxnMessage.PostInrlTxn.seTb.seEnt.seId":"CAKE","TxnMessage.PostInrlTxn.seTb.seEnt.rCd":678,"TxnMessage.HeaderData.EventCatg":"F","TxnMessage.PostInrlTxn.Key.Acnt":"1234567890","TxnMessage.PostInrlTxn.Key.Date":"20181018","TxnMessage.PostInrlTxn.Id":"3456","TxnMessage.PostInrlTxn.escTx":"Reload","TxnMessage.PostInrlTxn.Bal":6766} {"TxnMessage.PostInrlTxn.seTb.seEnt.seId":"","TxnMessage.PostInrlTxn.seTb.seEnt.rCd":0,"TxnMessage.HeaderData.EventCatg":"F","TxnMessage.PostInrlTxn.Key.Acnt":"1234567890","TxnMessage.PostInrlTxn.Key.Date":"20181018","TxnMessage.PostInrlTxn.Id":"3456","TxnMessage.PostInrlTxn.escTx":"Reload","TxnMessage.PostInrlTxn.Bal":6766} @Matt Burgess @Matt Clarke @Shu Thanks

ishantiwari91 · ‎04-26-2019

@Shu This solution doesn't work if I have multiple arrays in one json message. Example - {"TxnMessage": {"HeaderData": {"EventCatg": "F"},"PostInrlTxn": {"Key": {"Acnt": "1234567890","Date": "20181018"},"Id": "3456","AdDa": {"Area": [{"HgmId": "","HntAm": 0},{"HgmId": "","HntAm": 0}]},"escTx": "Reload","seTb": {"seEnt": [{"seId": "CAKE","rCd": 678},{"seId": "","rCd": 0}]},"Bal": 6766}}} If you could help with analysis and/or pointers on how each element of array in such input can be split into a separate rows?

ishantiwari91 · ‎04-04-2019

@Matt Clarke Thanks a lot. You are genius!

ishantiwari91 · ‎03-29-2019

@Matt Clarke Its a nifi cluster having 3 nodes. That's why I am getting 3 files. Is there any way to handle this? I have already I invested my 4 days into this. Lol.

ishantiwari91 · ‎03-29-2019

@Matt Clarke I have increased the Run Schedule to 2 mins and made sure that queue has all the flowfiles before executing MergeRecord processor. Still my output is 3 files of 4KB each but the desired outcome is one merged file. Not sure what configuration is impacting the result.

ishantiwari91 · ‎03-28-2019

We are receiving Json messages from upstream system via Kafka topic. Requirement is to store these messages into HDFS at certain interval. Since we are storing into HDFS we want to merge certain number of these Records in to single file. As per NiFi documentation we are using "MergeRecords" processor for that. About the in coming Records:## These are the multi-line JSon messages with nested structure. Those are based on the same schema (they are picked from single Kafka topic) Those are validated messages and even NiFi processor is able to parse it. so apparently no issues with JSon messages from Schema point of view Present Configuration Below is the snapshot of the Processor Configuration. NiFi version: 1.8 Expected behavior For the Above configuration its expected that MergeRecords should have weighted for one of the thresholds i.e. Maximum records(100000) or Maximum Bean size(100KBs). Observed Behavior But its observed that bean is getting bundled pretty before either of the threshold is reached. It is triggering the bean formation only for 2 records of 5KB size. If you could help with analysis and/or pointers as why MergeRecord processor is not behaving as per the configuration? @Matt Burgess @Matt Clarke

ishantiwari91 · ‎03-27-2019

I am trying to merge all json messages into a single file through custom groovy script. From MergeRecord I am not getting a single file rather I am getting small merged files. So as a workaround I am trying to write a groovy script for the same.

Online	Offline
Last Visited	‎07-19-2019 03:53 PM

Date Registered	‎03-22-2019 07:12 AM
Last Visited	‎07-19-2019 03:53 PM
Total Messages Posted	9

Re: Nifi denormalize json message and storing in H...

Nifi denormalize json message and storing in Hive

Re: Need to display each element of array in a sep...

Re: Nifi MergeRecord processor not merging as per ...

Re: Nifi MergeRecord processor not merging as per ...

Re: Nifi MergeRecord processor not merging as per ...

Nifi MergeRecord processor not merging as per conf...

Groovy script for merging flowfiles