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

@Shailesh Nookala Sure, if you are thinking to insert the data to Sql Server we can do that by using NiFi, You don’t have to download the file at all. You can do that couple of ways by using ExtractText Processor and extract all the contents of csv file into attributes and prepare insert statement by using ReplaceText Processor and push the data to Sql server. Another method is preparing json document for input CSV file and extracting all the attributes of flow file and prepare record to insert data to SQL server. I’m using Json method which is easy to extract attributes of flowfiles. Example:- Let’s consider you are having the csv file having just 2 records in it. 1,test,1000 2,test1,2000 As I’m using NiFi 1.1 there is no direct processor which can convert the incoming csv file to json. we need to follow InferAvroSchema--> ConvertCSVToAvro --> ConvertAvroToJSON the output would be in Json format with the same contents of our input CSV file. But in new versions of NiFi(I think from 1.3) there are processors to convert CSV to Json directly using ConvertRecord Processor. InferAvroSchema:- In this processor I’m giving the incoming csv file definition (or) you can read header definition from incoming flowfile by changing Get CSV Header Definition From Data property to true NiFi can read definition from 1 line of file. as I know my incoming file having id,name,salary I have given this definition and keep Schema Output Destination to flowfile-attribute we can make use of this schema in ConvertCSVToAvro processor. ConvertCSVToAvro:- change the Record Schema property to ${inferred.avro.schema} Which can infer the schema from flowfile attribute. ConvertAvroToJSON:- drag and drop ConvertAvroToJSON processor and leave the properties to default. Output:- [{"id": 1, "name": "test", "salary": 1000},{"id": 2, "name": "test1", "salary": 2000}] SplitJson:- If you are having more than 1 record in your csv file use splitjson processor because if ConvertAvroToJSON processors find more than one record in the flowfile it will make an array and keep all the records inside the array. As we need only 1 record in the flowfile before inserting data to Sqlserver. If your flowfile having array then json path expression property should be $.* as it will dynamically splits the array of records to one record for one flowfile. Input:- [{"id": 1, "name": "test","salary": 1000},{"id": 2, "name":"test1", "salary": 2000}] Output:- flowfile1:- {"id": 1, "name": "test", "salary": 1000}flowfile2:- {"id": 2, "name": "test1", "salary": 2000} As Splijson splits the array into individual records. EvaluateJsonPath configs:- in this processor extracting all the contents of flowfile to attributes by changing the list of properties Destination as flowfile-attribute extracting contents of json file by adding new properties by clicking + symbol on right corner. id as $.id name as $.name salary as $.salary ReplaceText Configs:- We are preparing insert statement in this processor and change Replacement Strategy property as AlwaysReplace Use insert statement and give destination table name and use the extracted attributes to replace the contents of values dynamically. insert into sqlserver_tablename (id,name,salary) values (${id},${name},${salary}) output:- flowfile1 content:- insert into sqlserver_tablename (id,name,salary) values(1,test,1000) flowfile2 content:- insert into sqlserver_tablename (id,name,salary) values(2,test1,2000) PutSQL Configs:- use putsql processor and make a connection pool to sql server enable and use the connection pool. so Putsql processor will executes all the insert statements that we are having contents of the flowfile and all the data will get inserted to sql server. FlowScreenshot:- ... View more

@Sami Ahmad, make sure you having you have given hive-site.xml,core-site.xml,hdfs-site.xml Paths in Hadoop Configuration Resources property and if you are using kerberos then you need to specify Kerberos Principal, Kerberos Keytab properties in PutHDFS processor. Example:- ... View more

@sally sally, can you make use of below search property ^<[^>]+>(.*)\<\/\?.*\>$ Replacetext Configs:- Input:- <?xml version="1.0" encoding="utf-8"?>abc</?xml version="1.0" encoding="utf-8"?> Output:- <DailyData>abc</DailyData> ... View more

Hi @sally sally, You can do that by using invokeHTTP processor, once you make first service call then keep Response relationship to trigger next service. This way we can only triggers next service once we get response from previous service. Example:- In my below flow service 1 is triggered by GenerateFlowFile processor then i connected response relationship to trigger service2 InvokeHTTP processor. This service2 processor only triggers when it got response from service1 processor and keep in mind the response from service1 will be overwritten by response of service2. ... View more