@manisha jain, Output port only transfers the data from a process group to Outside processors(or) processor groups. Can you add an Processor Group and use these two processors inside Processor Group then you can enable transmission from Output port. Flow:- Inside the processor group keep all your processors. ... View more

Adrian Oprea, Hi i have the same input json as you in generate flow file processor to test the entire flow, { "ARTGEntryJsonResult": { "AnnualChargeExemptWaverDate": "null", "Conditions": [ "" ], "ConsumerInformation": { "DocumentLink": "" }, "EntryType": "Medicine", "LicenceClass": "", "LicenceId": "152567" }, "Products": [ { "AdditionalInformation": [], "Components": [ { "DosageForm": "Drug delivery system, transdermal", "RouteOfAdministration": "Transdermal", "VisualIdentification": "Dull, homogenous" } ], "Containers": [ { "Closure": "", "Conditions": [ "Store at room temperature" ], "LifeTime": "2 Years", "Material": null, "Temperature": "Store below 25 degrees Celsius", "Type": "Sachet" } ], "EffectiveDate": "2017-09-18", "GMDNCode": "", "GMDNTerm": "", "Ingredients": [ { "Name": "Fentanyl", "Strength": "6.3000 mg" } ], "Name": "FENTANYL SANDOZ ", "Packs": [ { "PoisonSchedule": "(S8) Controlled Drug", "Size": "1" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "10" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "2" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "3" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "4" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "5" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "7" }, { "PoisonSchedule": "(S8) Controlled Drug", "Size": "8" } ], "SpecificIndications": [ "Management of chronic pain requiring opioid analgesia." ], "StandardIndications": [], "Type": "Single Medicine Product", "Warnings": [] } ] } JoltTransformation:- i use this processor because we need to extract the name attribute from products array [ { "operation": "shift", "spec": { "*": "&", "Products": { "*": "Products" } } } ] after this processor we get output like below:- { "ARTGEntryJsonResult" : { "AnnualChargeExemptWaverDate" : "null", "Conditions" : [ "" ], "ConsumerInformation" : { "DocumentLink" : "" }, "EntryType" : "Medicine", "LicenceClass" : "", "LicenceId" : "152567" }, "Products" : { "AdditionalInformation" : [ ], "Components" : [ { "DosageForm" : "Drug delivery system, transdermal", "RouteOfAdministration" : "Transdermal", "VisualIdentification" : "Dull, homogenous" } ], "Containers" : [ { "Closure" : "", "Conditions" : [ "Store at room temperature" ], "LifeTime" : "2 Years", "Material" : null, "Temperature" : "Store below 25 degrees Celsius", "Type" : "Sachet" } ], "EffectiveDate" : "2017-09-18", "GMDNCode" : "", "GMDNTerm" : "", "Ingredients" : [ { "Name" : "Fentanyl", "Strength" : "6.3000 mg" } ], "Name" : "FENTANYL SANDOZ ", "Packs" : [ { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "1" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "10" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "2" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "3" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "4" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "5" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "7" }, { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "8" } ], "SpecificIndications" : [ "Management of chronic pain requiring opioid analgesia." ], "StandardIndications" : [ ], "Type" : "Single Medicine Product", "Warnings" : [ ] } } without product as an array, now its easy to get Name attribute from the json message. Configs for jolt:- Evaluate Json path expression:- to extract licenseid and name attributes from the content i added the below properties as licenceid as $.ARTGEntryJsonResult.LicenceId name as $.Products.Name EJ configs:- Split Packs Array using splitjson processor:- change JsonPath Expression property to $.Products.Packs once you split the packs array then the every message in array will be one seperate flow file, as in my input we are having 8 messages in packs array so we get 8 flowfiles having licenceid,name attributes associated with each flowfile. Splijson Config:- Extract PoisonSchedule,Size using evaluatejson path:- now we need to extract all the contents of flowfile as attributes as we are having { "PoisonSchedule" : "(S8) Controlled Drug", "Size" : "1" } so we need to add 2 properties in processor and change Destination property to flowfile-attribute PoisonSchedule as $.PoisonSchedule Size as $.Size Configs:- Right now we are having all the desired contents of message as attributes we can use replace text processor(if you need output as text) (or) attributestojson processor(if you want json message) ReplaceText Processor:- we are having all the list of attributes ${licenseid},${name},${PoisonSchedule},${Size} here i kept , as seperator you can mention what ever you like, so keep them in Replacement Value property and change Replacement strategy to AlwaysReplace. Output:- flowfile1:- 152567,FENTANYL SANDOZ ,(S8) Controlled Drug,4 flowfile2:- 152567,FENTANYL SANDOZ ,(S8) Controlled Drug,3 Configs Replacetext:- AttributesToJSON:- if you want to convert the results as json documents then use this processor and in attributes list keep property as licenseid,name,PoisonSchedule,Size it will converts the attributes as json message Output:- {"Size":"10","PoisonSchedule":"(S8) Controlled Drug","name":"FENTANYL SANDOZ ","licenseid":"152567"} if you want to merge these flowfiles together,use Mergecontent Processor and change the properties as for your requirements. Flow Screenshot:- is attached in comments Hope this Helps...!! ... View more

Hi @Xtr yarhid, In Apache NiFi, Controller Services are shared services that can be used by Processors, i.e let's take if you are thinking to get the data or store the data to Hbase Hive tables then these processors need Hive,Hbase controller services First we needs to enable there services and then use them in the processors. Coming back to your question, Use ListHDFS processor this processor will store the state and run this processor for 15 mins using cron (or) timer driven if there is no changes made to the directory (or) file it won't list the flowfile, If there is any change in the directory or file then this processor gives only the new file that got changed in the directory and updates the state of processor with new file created timestamp Configure the directory property. In this way ListHDFS processor gives an flowfile with path and filename attributes which are used by FetchHDFS processor to fetch the data from HDFS directory. This processor won't do any fetching of files it will do just listing all the available files in the directory and FetchHDFS processor will do actual fetching of files. ListHDFS Configs:- in this processor i have given /user/yashu/del_test as the directory property and this process runs for every 900 sec on Timer driven, for the first time this processor lists all the files that are in del_test directory and filename,path will attributes with the flow file(if you are having 2 files in the directory then there will be 2 flowfiles each ff will have file name and path attribute to it). if you want to see the state in ListHDFS processor right click on processor and click on view state button. FetchHDFS:- Then use FetchHDFS processor and leave that with default configs as this processor gets attributes ${path}/${filename} from ListHDFS processor. in this processor fetches actual data from HDFS as list hdfs processor only lists the files that are changed in last 15 mins. In addition, this way after ListHDFS processor you can use Site-to-site processor, S2S will distributes the work across the cluster and use FetchHDFS we can do actual fetching the data. Hope this helps..!! ... View more

@sally sally Route on content is to match the contents of ff we cannot compare attribute inside this processor. But we can acheive by using RouteOnAttribute Processor:- as we are having ${message.body} as attribute then add a property ${message.body:contains('<person>')} it will look for message.body attribute contains <person> or not. if yes then it routes to person relation and if ff not having <person> it will routes to unmatched relation. RouteOnContent Processor:- As you are having $message.body as attribute we are not able to use RouteOnContent processor, because it will look the content of ff to find match. E.g:-How RouteOnContent works? Input:- <?xml version="1.0" encoding="UTF-8"?> <note> <person>Tove</to> <from>Jani</from> <heading>Reminder</heading> <body>Don't forget me this weekend!</body> </note> If we are having the above input as the ff content then in Route on content processor we need to give property as It will look for is the content of ff having <person> or not In our case it is yes then it will route to person relation. ... View more

Hi @Sanaz Janbakhsh Prepare INSERT/UPDATE statements:- As you are having array inside your json message,if you have array in your json file then you need to flatten out json message before ConvertJSONToSQL processor as it expects flat json message to prepare INSERT/UPDATE statements.I tried with your data There are 2 ways to prepare INSERT/UPDATE statements for PutSQL processor. Method1:- Using JSONToSQL to prepare INSERT/UPDATE:- Input:- { "applicationID": "9", "applicationName": "Sensors", "nodeName": "XXXX", "devEUI": "XX", "rxInfo": [ { "mac": "647fdafffe00416b", "time": "2017-07-19T21:39:35Z", "rssi": -1, "loRaSNR": XXX, "name": "SaroshGateway", "latitude": 51.0216123, "longitude": -114.039802, "altitude": 0 } ], "txInfo": { "frequency": 902300000, "dataRate": { "modulation": "LORA", "bandwidth": 125, "spreadFactor": 10 }, "adr": true, "codeRate": "4/5" }, "fCnt": 7, "fPort": 1, "data": "IA==" } Flattening Base Level Json data:- First we need to flatten out the base fields which are not in Array. To flatten Json message in nifi we are having JoltTransformJSON processor which will do flattening the json file. You can test out flattening json on http://jolt-demo.appspot.com Use the Jolt Specification below to flatten the base fields:- [ { "operation": "shift", "spec": { "*": "&", "txInfo": { "*": "tI-&", "dataRate": { "*": "dR&" } } } } ] Output:- { "applicationID": "9", "applicationName": "Sensors", "nodeName": "XXXX", "devEUI": "XX", "rxInfo": [{ "mac": "647fdafffe00416b", "time": "2017-07-19T21:39:35Z", "rssi": -1, "loRaSNR": "XXX", "name": "SaroshGateway", "latitude": 51.0216123, "longitude": -114.039802, "altitude": 0 }], "tI-frequency": 902300000,--there is tI before frequency because in jolt i have given "*": "tI-&" "dRmodulation": "LORA",--same with dR also i have given "*": "dR&" "dRbandwidth": 125, "dRspreadFactor": 10, "tI-adr": true, "tI-codeRate": "4/5", "fCnt": 7, "fPort": 1, "data": "IA==" } In this step we have flatten all the base fields in json message except array. jolt configs:- Extract Attributes for Flatten Json Fileds using EvaluateJsonPath:- In this step use the Destination property as flowfile-attribute so that we can extract the content of the json message to attributes. Give all the attribute names which are not part of array "rxInfo": [{ "mac": "647fdafffe00416b", "time": "2017-07-19T21:39:35Z", "rssi": -1, "loRaSNR": "XXX", "name": "SaroshGateway", "latitude": 51.0216123, "longitude": -114.039802, "altitude": 0 }] For all the other data in json message add properties to the by clicking + symbol on top right Add properties like below applicationID as $. applicationID Do the same thing for all the other attributes except for array elements. EvaluateJson configs:- Split Array using SplitJson Processor:- We are having array which we haven’t transformed yet so use splitjson processor and property for JsonPath Expression as $.rxInfo. It will splits the json array and flatten json message we are having as a flowfile now. Once we got flatten json message extract all the values of message to attributes by using evaluate json path processor as we did before. AttributesToJSON Processor: For every flowfile we are having all your json message data as attributes now so use Attributestojson processor to get all the attributes into json message as this processor will result flatten json message. In AttributesList property give all the list of attributes that you are thinking to make them as json message. nodeName,fPort,data,fCnt,devEUI,ri_altitude,tI-frequency,applicationID,applicationName,ri_loRaSNR,ri_time,ri_name,dRbandwidth,ri_latitude,dRmodulation,dRspreadFactor,tI-codeRate,tI-adr,ri_mac,ri_rssi,ri_longitude Output:- { "nodeName": "XXXX", "fPort": "1", "data": "IA==", "fCnt": "7", "ri_loRaSNR": "XXX", "devEUI": "XX", "ri_time": "2017-07-19T21:39:35Z", "ri_name": "SaroshGateway", "dRbandwidth": "125", "ri_latitude": "51.0216123", "dRmodulation": "LORA", "dRspreadFactor": "10", "ri_altitude": "0", "tI-frequency": "902300000", "tI-codeRate": "4/5", "tI-adr": "true", "ri_mac": "647fdafffe00416b", "applicationID": "9", "ri_rssi": "-1", "applicationName": "Sensors", "ri_longitude": "-114.039802" } ConvertJSONToSQL processor:- Right now we are having flatten json message we can make use JSONToSQL processor to prepare INSERT/UPDATE statements, as keep in mind I have renamed the array and some other message data you can change the attribute value in evaluate json path (or) in jolt transform processors. Method2:- Replace text to prepare INSERT/UPDATE statements As my flow screenshot shows Use Replace text processor after you got all the message contents as flowfile attributes(i.e after Extract Attributes for Json Array data) processor Replace text Processor Configs:- We are have all the contents of json message to attributes for each flow file so use those attributes inside replace text processor and change the following properties Replacement Value property to INSERT INTO table_name (nodeName,fPort,....) VALUES (${nodeName},${fPort},....) Here we are preparing insert sql statement and replacing with attribute values as ${nodeName} returns the value associated with the flowfile. Output:- INSERT INTO table_name (nodeName,fPort,....) VALUES (XXXX, 1,....) ReplaceText Config:- Right now we have prepared insert and update statements and replaced all the attributes with values associated with the ff. PutSQL:- From both methods will prepare INSERT/UPDATE statements then use PutSQL processor and Configure the connection pool and auto terminate success message. Flow screenshot:- Take reference as my screenshot and make sure you have given Correct Relations(like success,split,matched...) to the next processors. Hope this helps..! ... View more

Hi @Biswajit Chakrabort, As you are having rolling logs for daily, i tried to TailFile processor by using File(s) to Tail property as follows /my/path/directory/my-app-${now():format("yyyy-MM-dd")}.log The above expression looks for my-app-2017-09-24.log file in /my/path/directory and tails the file if the file is presented in that directory. ... View more

Hi @Ravi Koppula, I don't think there is a way to schedule processor groups but there is a way to schedule them all of them on one shot. 1.As you are using ExecuteStreamCommand processor,processor will accepts incoming connections as we can schedule the processor in Timer driven,Cron driven,Event Driven. 2.Just use GenerateFlowfile processor and give success relation to all ExecuteStreamCommand processors and schedule GenerateFlowfile to schedule a run on either cron (or) timer driven. 3.Generateflowfile processor runs on scheduled time and gives trigger flowfile to run all the other ExecuteStreamCommand processors. Sample Flow:- ... View more