Hi @xav webmaster, Use scancontent processor with DictionaryFile needs to be in all nifi nodes, in my case my file is in /tmp cat dict.txt ford So this processor looks into content and if it matched with the dictionary file then adds matching.term attribute to the flowfile. If your content of the flowfile is multiple lines then it will looks for all the lines if ford word is there then it adds matching.term attribute to the flowfile. Once you got matching.term attribute then use update attribute processor to change the name of the attribute (or) you can use same attribute. add new property kafka_topic ${matching.term} Configs:- after this processor matching.term attribute renamed to kafka_topic Another way to achieve this case is 1.ExtractText to get contents to attributes 2.UpdateAtrribute and add advance properties to check if the attribute extracted before having the required content by using NiFi expression Language 3. you will get same result as you got using scan content processor. ... View more

Hi @rakesh chow, as ExecuteScripts are experimental in NiFi but you can acheive same results by using another processors as follows. 1.EvaluateJsonPath //Extract all the contents to attributes 2.UpdateAttribute //Update time stamps attributes with new format(yyyy-mm-dd hh...etc) 3.AttributesToJson //Recreate Json content by specifying all the attributes EvaluateJsonPath configs:- My input sample content is { "ID":"1", "CID":"1", "DiscoveredTime":"Mon Sep 11 19:56:13 IST 2017", "LastDiscoveredTime":"Mon Oct 09 23:38:55 IST 2017" } in eval json processor extract all the contents of the flowfile to the attributes of flowfile by changing Destination property to flowfile-attribute and add new properties by clicking + sign at right corner. ID $.ID DiscoveredTime $.DiscoveredTime CId $.CId LastDiscoveredTime $.LastDiscoveredTime so i have extracted all the content of ff as attributes of ff, in your case you need to add all your json content keys of contents here and this processor keeps all the values of content to attributes. Config Screenshot:- UpdateAttribute Processor Configs:- This processor will helps to update the attributes on the fly we are changing our DiscoveredTime,LastDiscoveredTime attributes on the fly. add new properties 1.DiscoveredTime ${DiscoveredTime:toDate('EEE MMM dd HH:mm:ss ZZZ yyyy'):toNumber():plus(21600000):format("yyyy-MM-dd HH:mm:ss.SSS")} in this expression we are checking which timezone it is and converting that to number and adding 6 hrs(i think you don't need to do :plus(21600000)), after that i'm converting to yyyy-MM-dd HH:mm:ss.SSS format Your expression probably will be ${DiscoveredTime:toDate('EEE MMM dd HH:mm:ss ZZZ yyyy'):toNumber():format("yyyy-MM-dd HH:mm:ss.SSS")} 2. add new property for LastDiscoveredTime ${LastDiscoveredTime:toDate('EEE MMM dd HH:mm:ss ZZZ yyyy'):toNumber():plus(21600000):format("yyyy-MM-dd HH:mm:ss.SSS")} Your expression:- ${LastDiscoveredTime:toDate('EEE MMM dd HH:mm:ss ZZZ yyyy'):toNumber():format("yyyy-MM-dd HH:mm:ss.SSS")} Configs:- AttributesToJson Processor:- in Attributes list property add all the available attributes it will prepare a json messages with the attributes that you have mentioned here. ID,CID,DiscoveredTime,LastDiscoveredTime Change Destination property to flowfile-content Configs:- Input:- { "ID":"1", "CID":"1", "DiscoveredTime":"Mon Sep 11 19:56:13 IST 2017", "LastDiscoveredTime":"Mon Oct 09 23:38:55 IST 2017" } Output:- { "ID" : "1", "LastDiscoveredTime" : "2017-10-09 23:38:55.000", "DiscoveredTime" : "2017-09-11 19:56:13.000", "CID" : "" } In your case you need to mention all your attributes in attributes to json processor and it will prepare new json content with all your attributes, if you left this property as empty then all the attributes associated with the ff will be part of your json content. Flow Screenshot:- ... View more

@Kiem Nguyen you can use RouteonContent processor and change Match Requirement property to Content must contain Match and add the below properties 1.project_1 as .*project_1.* 2.project_2 as .*project_2.* This processor will checks the content of flowfile if it is having project_1 in it it transfers the ff to project_1 relationship, if the content having project_2 then it transfers to project_2 relationship. Flow Screenshot:- As shown in the above screenshot you can use the relationships project_1,project_2 and store or process those content as per your requirements. ... View more

@Ramya Jayathirtha, As i'm having id,name,age columns in foo table when ever we does Hive# select name from foo; //in this case first map phase will loads the file and we only selected name column, we are not doing any filtering kind of things here so map phase checks name field and gives results. MapSideJoins:- Usually all joins will perform on reducer side as we can explicitly mention load tables to memory and performs joins, no reducer phase will be initialized. Hive# select /*+MAPJOIN(..)*/... //this kind of joins will loads small table to memory and does the join on map phase only. When ever we do insert values into table and loading the data should be used only map phase. Hive# insert into foo values(1,'abc',200); INFO : Map 1: -/- INFO : Map 1: 0/1 INFO : Map 1: 0(+1)/1 INFO : Map 1: 1/1 INFO : Table default.foo stats: [numFiles=5, numRows=5, totalSize=38, rawDataSize=33] Simple CTAS without Aggregations:- When we does Create table as simple select then only mapper phase will be initialized. if we does any aggregations then reducer phase will get initialized Hive#create table foo1 stored as orc as select * from foo INFO : Map 1: -/- INFO : Map 1: 0/1 INFO : Map 1: 0(+1)/1 INFO : Map 1: 1/1 INFO : Table default.foo1 stats: [numFiles=1, numRows=4, totalSize=XXX, rawDataSize=XXXX] No rows affected (10.247 seconds) Hive#select * from foo1; +----------+------------+-----------+--+ | foo1.id | foo1.name | foo1.age | +----------+------------+-----------+--+ | 1 | a | 10 | | 2 | a | 11 | | 2 | a | 10 | | 3 | b | 10 | | 4 | b | 10 | | 5 | c | 10 | +----------+------------+-----------+--+ 6 rows selected (0.205 seconds) 2. if we does CTAS with where clause in it still it is just map phase all the filters in WHERE clause are going to be done by mapper phase it self. Hive#create table foo as select * from foo1 where id='1'; INFO : Map 1: -/- INFO : Map 1: 0/1 INFO : Map 1: 0(+1)/1 INFO : Map 1: 1/1 INFO : Table default.foo stats: [numFiles=1, numRows=1, totalSize=7, rawDataSize=6] No rows affected (9.984 seconds) Hive#SELECT * FROM FOO; +---------+-----------+----------+--+ | foo.id | foo.name | foo.age | +---------+-----------+----------+--+ | 1 | a | 10 | +---------+-----------+----------+--+ 1 row selected (0.099 seconds) ... View more

@xav webmaster We can do that by using RouteonContent processor followed by UpdateAttribute processor. Routeoncontent:- change Match Requirement strategy to content must contain match add the properties adult_dog as (.*adult.*dog.*) cat_dog as (.*cat.*dog.*) Config:- Right now we are checking the contents of flowfile and add new properties for all the cases that you need to publish the data to kafka, then use UpdateAttribute processor: Add your deserved name to kafka_topic attribute of the flowfile. add new property kafka_topic as cat_dog Configs:- PublishKafka:- every flowfile will have only kafka_topic as the attribute having different topic names in it, make use of kafka_topic in our publish kafka processor. Change Topic Name property to ${kafka_topic} Configs:- Flow screenshot:- Flow Explanation:- Routeoncontent //check the content and transfers flowfile to matching relationships. UpdataAttribute //adding kafka_topic attribute with some name. PublishKafka //using kafka_topic attribute and publishing the contents to respective topics dynamically. Like this way we don't have to use any scripts and dynamically publish into kafka topics. ... View more