Hello, I desinged the flow according to this link(https://community.hortonworks.com/questions/189153/reading-an-email-using-apache-nifi-and-extracting.html). Using this flow i got all the attribute in file and its working perfectly. Consume Imap processors --> ExtractEmailHeader Processor --> Putfile Processor. But i want to sort the specific attribute from the flow file for this when i m setting the attribute value in replaceText processor that time getting the attribute value. email.headers.sent_date :${email.headers.sent_date} email.headers.subject :${email.headers.subject} email.attachment_count :${email.attachment_count} how i will get(set) the value of that type of variable using expression language in Apache NIFI email.headers.bcc.* Each individual BCC recipient (if available) email.headers.cc.* Each individual CC recipient (if available) email.headers.from.* Each individual mailbox contained in the From of the Email (array as per RFC-2822) My designed flow Consume Imap processors --> ExtractEmailHeader Processor --> replaceText processor --> Putfile Processor. Thanks in advance @Shu @jwitt ... View more

Hello, I am trying to convert multiple json file into csv file in that, In source folder i am getting different json file with different schema. need to convert json file to csv file with dynamic schema mapping and put that file into destination folder. I have tried bellow flow, 1) getFile processor- to get file from source. 2) UpdateAttribute - set attribute schema.name : record 3) inferAvroSchema -configure Schema Output Destination : flowfile-attribute Input Content Type : json 4)convertRecord processor - Record Reader :JsonTreeReader Record Writer :CSVRecordSetWriter 5) putFile : put file into destination folder. But in above flow getting error in convertRecord Processor, Error: Failed to process StandardFlowFileRecord[uuid=69603ac0-ac43-44b5-82f3-5c1c89e4abaf,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1547526092906-1, container=default, section=1], offset=878, length=878],offset=0,name=orders1.json.avro,size=878]; will route to failure: Failed to create schema from the Schema Text after evaluating FlowFile Attributes. Please find the attached screen-shot of configuration of convert record processor. Thanks ... View more

Hello, In file watcher scenario, everyday i am getting new and updated file in the local folder. The incoming file count is varies in everyday. Need to send email notification with updated and new file name list with specific schedule time. Using Listfile processor, I am getting all new and updated file name. How to list out all filename in one file and send that file using email notification. Thanks ... View more

Hello, 1) I am fetching data from relational database and then convert that data into flat json. 2)split json file using splitJson processor and then convert json file into nested json(that part working properly). Input json {"RqUID":"UID.1","RqCreatedDt":"2017-06-22T13:10:03Z","RqMsgUID":"Kp-gp1-acc-1.3","RqType":"ACCTTRNHISTADD","RqMsgCreatedDt":"2017-06-22T13:10:03Z","AcctTrnId":"Kp-gp1-acc-1-trn.4","TrnType":"POSTED SALE","TrnSrc":"PURCHASE - AT TERMINAL","MCC":"4112","Memo":"commuter","Amt":"180000.0","AmtType":"DR","CurCode":"IDR","FXTrnInd":"false","AcctId":"Kp-gp1-acc-1","AcctType":"CCA","StatusCode":"POSTED","OrigDt":"2018-04-26T04:00:00.100Z"} Jolt Specification [ { "operation": "shift", "spec": { "RqUID": "RqHdr.RqUID", "RqCreatedDt": "RqHdr.RqCreatedDt", "RqMsgUID": "RqMsg[0].RqMsgUID", "RqType": "RqMsg[0].RqType", "RqMsgCreatedDt": "RqMsg[0].RqMsgCreatedDt", "AcctTrnId": "RqMsg[0].AcctTrnRec.AcctTrnId", "TrnType": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnDetails.TrnType", "TrnSrc": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnDetails.TrnSrc", "MCC": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnDetails.MCC", "Memo": "RqMsg[0].AcctTrnRec.AcctTrnInfo.Memo", "Amt": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnAmtInfo.Amt", "AmtType": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnAmtInfo.AmtType", "CurCode": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnAmtInfo.CurCode", "FXTrnInd": "RqMsg[0].AcctTrnRec.AcctTrnInfo.FXTrnInd", "AcctId": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnAcctInfo.AcctId", "AcctType": "RqMsg[0].AcctTrnRec.AcctTrnInfo.TrnAcctInfo.AcctType", "StatusCode": "RqMsg[0].AcctTrnRec.AcctTrnStatus.StatusCode", "OrigDt": "RqMsg[0].AcctTrnRec.AcctTrnStatus.OrigDt" } } ] output json {"RqHdr":{"RqUID":"UID.1","RqCreatedDt":"2017-06-22T13:10:03Z"},"RqMsg":[{"RqMsgUID":"Kp-gp1-acc-1.3","RqType":"ACCTTRNHISTADD","RqMsgCreatedDt":"2017-06-22T13:10:03Z","AcctTrnRec":{"AcctTrnId":"Kp-gp1-acc-1-trn.4","AcctTrnInfo":{"TrnDetails":{"TrnType":"POSTED SALE","TrnSrc":"PURCHASE - AT TERMINAL","MCC":"4112"},"Memo":"commuter","TrnAmtInfo":{"Amt":"180000.0","AmtType":"DR","CurCode":"IDR"},"FXTrnInd":"false","TrnAcctInfo":{"AcctId":"Kp-gp1-acc-1","AcctType":"CCA"}},"AcctTrnStatus":{"StatusCode":"POSTED","OrigDt":"2018-04-26T04:00:00.100Z"}}}]} 3) Problem I want direct convert flat array json into the nested json one show as bellow a) when trying to convert array of flat json file into nested json array file,i am getting output null nested array flat input json [{"RqUID":"UID.1","RqCreatedDt":"2017-06-22T13:10:03Z","RqMsgUID":"Kp-gp1-acc-1.3","RqType":"ACCTTRNHISTADD","RqMsgCreatedDt":"2017-06-22T13:10:03Z","AcctTrnId":"Kp-gp1-acc-1-trn.4","TrnType":"POSTED SALE","TrnSrc":"PURCHASE - AT TERMINAL","MCC":"4112","Memo":"commuter","Amt":"180000.0","AmtType":"DR","CurCode":"IDR","FXTrnInd":"false","AcctId":"Kp-gp1-acc-1","AcctType":"CCA","StatusCode":"POSTED","OrigDt":"2018-04-26T04:00:00.100Z"},{"RqUID":"UID.1.kp.gp1-acc-1.01","RqCreatedDt":"2017-06-22T13:10:03Z","RqMsgUID":"Kp-gp1-acc-1.3","RqType":"ACCTTRNHISTADD","RqMsgCreatedDt":"2017-06-22T13:10:03Z","AcctTrnId":"Kp-gp1-acc-1-trn.4","TrnType":"POSTED SALE","TrnSrc":"PURCHASE - AT TERMINAL","MCC":"4112","Memo":"commuter","Amt":"180000.0","AmtType":"DR","CurCode":"IDR","FXTrnInd":"false","AcctId":"Kp-gp1-acc-1","AcctType":"CCA","StatusCode":"POSTED","OrigDt":"2018-04-26T04:00:00.100Z"}] Expected Output {"RqHdr":{"RqUID":"UID.1","RqCreatedDt":"2017-06-22T13:10:03Z"},"RqMsg":[{"RqMsgUID":"Kp-gp1-acc-1.3","RqType":"ACCTTRNHISTADD","RqMsgCreatedDt":"2017-06-22T13:10:03Z","AcctTrnRec":{"AcctTrnId":"Kp-gp1-acc-1-trn.4","AcctTrnInfo":{"TrnDetails":{"TrnType":"POSTED SALE","TrnSrc":"PURCHASE - AT TERMINAL","MCC":"4112"},"Memo":"commuter","TrnAmtInfo":{"Amt":"180000.0","AmtType":"DR","CurCode":"IDR"},"FXTrnInd":"false","TrnAcctInfo":{"AcctId":"Kp-gp1-acc-1","AcctType":"CCA"}},"AcctTrnStatus":{"StatusCode":"POSTED","OrigDt":"2018-04-26T04:00:00.100Z"}}},{"RqMsgUID":"Kp-gp1-acc-1.3","RqType":"ACCTTRNHISTADD","RqMsgCreatedDt":"2017-06-22T13:10:03Z","AcctTrnRec":{"AcctTrnId":"Kp-gp1-acc-1-trn.4","AcctTrnInfo":{"TrnDetails":{"TrnType":"POSTED SALE","TrnSrc":"PURCHASE - AT TERMINAL","MCC":"4112"},"Memo":"commuter","TrnAmtInfo":{"Amt":"180000.0","AmtType":"DR","CurCode":"IDR"},"FXTrnInd":"false","TrnAcctInfo":{"AcctId":"Kp-gp1-acc-1","AcctType":"CCA"}},"AcctTrnStatus":{"StatusCode":"POSTED","OrigDt":"2018-04-26T04:00:00.100Z"}}}]} Trying from long time but not getting exact solution, can some one help. Thanks @Shu @Gonçalo Cunha ... View more

Hi, H have created flow, 1) Get Json file using getfile processor and post that file using Invokehttp processor(https://integration-laven-ie-prod.laven.com/lm/test/). 2) I have 3 files laven-ie-prod-laven-integration.crt laven-ie-prod-laven-integration.key laven-ie-prod-laven-integration.p12 3) Can someone help me how to use that file in Invokehttp processor, a)When i am using .crt and .key file that time StandardSSLContextService show keystores properties is invalid b)When using .p12 file that time show error Routing to Failure due to exception: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target: javax.net.ssl.SSLHandshakeException: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target can someone help me to figure out this issue. Thanks in advance. ... View more

Hello, I have to convert my flat json file into nested one. I have attached the flat json input file and nested json output file. Any help with JOLT Chain spec is much appreciated. Thanks flat-jsoninputfile.txtnested-outputfile.txt ... View more

Hello, I am getting files in batches in my local folder like 1st batch getting 4 files(a.txt, b.log, c.error, d.csv) After getting all files then after files will transfer to hdfs. If any files fails, the files will not transfer to hdfs and send email notification to user. Same for 2nd batch getting two files. Any ideas or suggestions... Thanks! ... View more

Hello, I have a two flow, from every flow get one flow file. May be there some delay between getting file from different flow How to compare one flow file attribute with other flow file attribute. After that send mail depending on success or failure condition Is there any easy way to do that in Apache NIFI. Thanks in advance. ... View more

Hello, I have Fetch Open source API data using GetHTTP processor and save as text file. Is there any way to auto increment page because size of data is more. Thanks ... View more

Hello, I am new in Apache NIFI, I have to import CRM data into hive. what is possible way to import data into hive. Thanks in advance ... View more

Hello, I have designed data flow to fetch data from twitter and put in kafka. Getting data in json format but not user specific. 1) That processor stream live Twitter data or old data? 2) If processor stream live data then how to stream old data ? 3) How to fetch user specific incremental data using gettwitter processor ? 4) If my flow failed some times then how to stream that specific time and live data using gettwitter processor ? Thanks. ... View more

I have design data flow to fetch data from twitter and put in kafka, getting data in json format but not user specific. 1) That processor stream live data or old data ? 2) if processor stream live data then how to stream old data ? 3) how to fetch user specific incremental data using gettwitter processor ? 4) if my flow failed some times then how to stream that specific time and live data using gettwitter processor ? Thanks, Mitthu ... View more

Hello, I have created a flow like ExecuteSQL->putfile on success relation if ExecuteSQL processor failed I have to capture the error of ExecuteSQL processor and send email notification to user with error file Thanks ... View more

Hello, I have design the flow in that I executed the query like CTAS,Insert into, truncate table in PutHiveQL Processor. Depending on query failed or success I have to send email but PutHiveQL not generated any metadata. I have set rollback property true but that data goes on queue. How to send email depending upon PutHiveQL Processor success or failure. ... View more

Hello, I have design the nifi flow like this selectHiveQL->convertAvroToJson->SplitJson->EvaluateJsonPath->RoutOnAttribute(depends on condition the query goes on) ->presto or ->ReplaceText->putHiveQL hive table is query_id job_type query 1 presto create table hive.db.table as select * from mysql.db.table 2 hive ALTER table db.table set tblproperties ("orc.create.index"="true") 3 hive ALTER table db.table set tblproperties ("transactional"="true") 4 presto create table hive.db.table_stage as select * from mysql.db.table selectHiveQL-select query on hive table contain query_id, job_type and query convertAvroToJson->convert avro format into json SplitJson-split the record EvaluateJsonPath-set the query_id, job_type and query attribute RoutOnAttribute-depending on condition of job_type that query is executed in presto or hive presto-custom processor executed the query ReplaceText-set the putHiveQL query putHiveQL-execute hive query complete flow-flow.png problems: 1) How to stop the flow if any query failed from any processor. 2) I have tried to connect the presto failure to presto processor that time if query failed the next query not goes to presto processor but failed query is continuously executed in presto processor. 3) if the query is alternate and presto query failed that time hive query goes to hive processor(backpressure of hive failure also set to 1). 4) How to send email notification if the flow is successfully completed or failed? Please provide your suggestions . Thanks in advance . ... View more

Hello, I have created a custom processor same as mysql , In that fetch queryID,query Start time,query end time from resultset and added that value in flowfile and flow file output obtained in json format, [{"r_regionkey":"0","r_name":"AFRICA","r_comment":"lar deposits. blithely final packages cajole. regular waters are final requests. "}, {"r_regionkey":"1","r_name":"AMERICA ","r_comment":"hs use ironic, even requests. s"}, {"r_regionkey":"2","r_name":"ASIA ","r_comment":"ges. thinly even pinto beans ca"}, {"r_regionkey":"3","r_name":"EUROPE","r_comment":"ly final courts cajole furiously final excuse"}, {"r_regionkey":"4","r_name":"MIDDLE EAST","r_comment":"uickly special accounts cajole carefully blithely close requests."}, {"Query_ID":"20171006_044330_00004_qtmrh","Query_StartTime":"2017-10-06 10:13:30.516","Query_EndTime":"2017-10-06 10:13:30.646"}] I have to insert QueryID, Query_StartTime, Query_EndTime in mysql table. Can someone please point me in the right direction how to do that. ... View more

I have custom processor like ExecuteSql . after executing query i get resultset. now i want add new values like queryid , starttime , and end time in resultset and send in flow file like normal result set. How i can do this. please advice thank you in advance ... View more