@Abhinav Joshi As you are Fetching results of Join in this case, We need to store the state in DistributeMapCache/HDFS/Hive/Hbase and then pull the state value from DistributeMapCache/HDFS/Hive/Hbase and use that value as lower bound, run the join incrementally. Example:- As i'm having emp,dept tables with joindate(timestamp datatype) as incremental column name in emp table. Flow:- Flow Explanation:- 1.GenerateFlowFile processor:- This processor will be trigger for the flow and Added a new property increment.value cache_key 2.FetchDistributeMapCache:- Configure and enable DistributedMapCacheClientService and change the below property values Cache Entry Identifier ${increment.value} Distributed Cache Service DistributedMapCacheClientService Put Cache Value In Attribute stored.state //if found then the value will be added to the flowfile as a attribute with name as stored.state 3.UpdateAttribute:- We have feeded both success/not-found relations to Update Attribute processor Add two properties current.state ${now():format("yyyy-MM-dd HH:mm:ss")} //current timestamp as current.state attribute stored.state ${stored.state:isNull():ifElse('1900-01-01 12:00:00','${stored.state}')} //if the value is null(in first run) then we are changing as 1900-01-01 12:00:00 and if the value is presented then we are keeping as is. You can change the default value, stored/current.state values as per your requirements. 4.ExecuteSql:- SQL select query select e.id,d.name from emp e join dept d on e.deptid=d.deptid where e.joindate > '${stored.state}' and e.joindate > '${current.state}' as my emp table having joindate as incremental column so i have used stored.state and current.state attribute values in my join where clause to run incrementally. -- Fork the success relation from executesql processor Fork1: to stored the current.state in distributecache Fork2: For other processing -- Fork1:To store the state:- 5.Replace Text:- Search Value (?s)(^.*$) Replacement Value ${current.state} Replacement Strategy Always Replace Evaluation Mode Entire text Now we are changing the flowfile content with current.state value as new flowfile content. - 6.PutDistributeCacheMap:- Cache Entry Identifier ${increment.value} - Fork2:-Further processing like processing (or) storing the results to HDFS..etc I have attached sample template below save/upload template to your NiFi instancce and change the configs as per your requirements. executesql-storestate-187764.xml ... View more

@Siddarth Wardhan Probably it's a bug in Hive ODBC driver, As a workaround you can use the below query, this query will run in beeline and in ODBC driver also. INSERT INTO Target_table(col_1, col_2, col_3) SELECT col_1, col_2,int(null) col_3 FROM Source_table; (or) INSERT INTO Target_table SELECT col_1, col_2,int(null) col_3 FROM Source_table; //if the order of columns matches between select statement and target_table These both query will insert same results as your query does (i.e . INSERT INTO Target_table(col_1, col_2) SELECT col_1, col_2 FROM Source_table;) ... View more

@Suresh Dendukuri One way of doing is to use Replace text processor to search for "NULL" in json record key/value and replace with empty string(blank), by using this way we can replace all NULL key/value pairs from flowfile content. Method1:- Input:- [ { "color": "black", "category": "hue", "type": "NULL", "code": { "rgba": [255,255,255,1], "hex": "#000" } }, { "color": "NULL", "category": "value", "code": { "rgba": [0,0,0,1], "hex": "#FFF" } }, { "color": "red", "category": "hue", "type": "primary", "code": { "rgba": NULL, "hex": "#FF0" } }] ReplaceText Configs:- Search Value ("type": "NULL",|"color": "NULL",|"rgba": NULL,) //search for all possible NULL key/value pairs(| is used as OR) Replacement Value //replace the with empty Character Set UTF-8 Maximum Buffer Size 1 MB //needs to change the size if your flowfile is more than 1 MB Replacement Strategy Regex Replace Output From Replace Text Processor:- [ { "color": "black", "category": "hue", "code": { "rgba": [255,255,255,1], "hex": "#000" } }, { "category": "value", "code": { "rgba": [0,0,0,1], "hex": "#FFF" } }, { "color": "red", "category": "hue", "type": "primary", "code": { "hex": "#FF0" } }] All NULL values are replaced from the Flowfile Content. (OR) Method2:- This way we are using ConvertRecord processor with JsonTreeReader as Record Reader and JsonSetWriter as Record Writer with the below property as Suppress Null Values Always Suppress //if the record having null value then we are not writing that into Output Flowfile. Flow:- GenerateFlowFile:- This processor is used to generate sample records Custom text property as [ { "color": "black", "category": "hue", "type": "NULL", "code": { "rgba": [255,255,255,1], "hex": "#000" } }, { "color": "NULL", "category": "value", "code": { "rgba": [0,0,0,1], "hex": "#FFF" } }, { "color": "red", "category": "hue", "type": "primary", "code": { "rgba": NULL, "hex": "#FF0" } }] add new property schema.name sch ReplaceText:- As we are going to use ConvertRecord processor and your input file having "NULL",NULL as values ,but these are not treated as null in Avro Schema Registry. so we are going to preparing json records which are going to understand by ConvertRecord Processor. --- Search Value (NULL|"NULL") //search for NULL (or) "NULL" in the flowfile content Replacement Value null //replace with null Maximum Buffer Size 1 MB Replacement Strategy Regex Replace Evaluation Mode Entire text Output:- [ { "color": "black", "category": "hue", "type": null, "code": { "rgba": [255,255,255,1], "hex": "#000" } }, { "color": null, "category": "value", "code": { "rgba": [0,0,0,1], "hex": "#FFF" } }, { "color": "red", "category": "hue", "type": "primary", "code": { "rgba": null, "hex": "#FF0" } }] All the "NULL",NULL and replaced with null in the flowfile content. ConvertRecord Processor:- Record Reader/Writer controller services are configured and enable the controller service and add AvroSchemaRegistry as we need to configure the avro schema with can match your input json record. JsontreeReader Configs:- ArrayAvroSchemaRegistry Configs:- sch { "type" : "record", "name" : "schema", "namespace" : "avroschema", "fields" : [ { "name" : "color", "type" : ["null","string"] }, { "name" : "category", "type" : "string" }, { "name" : "type", "type" : ["null","string"] }, { "name" : "code", "type" : { "type" : "record", "name" : "code", "fields" : [ { "name" : "rgba", "type" : { "type" : "array", "items" : ["null","int"] } }, { "name" : "hex", "type" : ["null","string"] }] } } ] } JsonSetWriter Configs:- Change the Supress Null Values to Always Suppress(we are not writing the null values for the json key's in the record) Output:- [{"color":"black","category":"hue","code":{"rgba":[255,255,255,1],"hex":"#000"}},{"category":"value","code":{"rgba":[0,0,0,1],"hex":"#FFF"}},{"color":"red","category":"hue","type":"primary","code":{"rgba":[],"hex":"#FF0"}}] all the null values in the json records are not written in the output flowfile but the only issues is with rgba array even we are having null values for the array still jsonsetwriter writer an empty array for rgba ReplaceText:- Now we are replacing the empty rgba empty array in this processor Search Value "rgba":[], //search for this value Replacement Value //same as method1 replacetext configs Maximum Buffer Size 1 MB Replacement Strategy Literal Replace Evaluation Mode Entire text Output:- [{"color":"black","category":"hue","code":{"rgba":[255,255,255,1],"hex":"#000"}},{"category":"value","code":{"rgba":[0,0,0,1],"hex":"#FFF"}},{"color":"red","category":"hue","type":"primary","code":{"hex":"#FF0"}}] I have attached my template below, save/upload the template and change as per your needs. remove-nulls-187756.xml - If the Answer Addressed Your Question, Click on Accept button below to accept the answer, That would be great help to Community users to find solution quickly for these kind of issues. ... View more

@Santanu Ghosh Q1: What is the correct way to define default Hive table ? It depends on how you datafile is delimited if your datafile is delimited with ^A(ctrl-a) then you don’t need to specify any delimiter,because hive default delimiter is ^A(ctrl-a) in ASCII. in this case if we do desc formatted <table-name>; hive doesn't shows any delimiter because table is created with default delimiter i.e ^A If your datafile is comma delimited then we need to create table with row format delimited fields terminated by ','. when we created table with create table mydb.user (uid int,name string); this ddl statement without any format and delimiters then hive creates user table with default serde (serialize,deserializer ). This serde instructs hive on how to process a record (Row) and serde library is inbuilt to Hadoop API. Default serde --> LazySimpleSerDe<br>Default InputFormat --> TextInputFormat<br>Default OutputFormat --> HiveIgnoreKeyTextOutputFormat<br>Default Delimiter --> ^A(ctrl-a) //although the delimiter is not showing in desc formatted <table-name>; Q2: Also, what is the default delimiter for Hive table ? By default, tables are assumed to be of text input format and the delimiters are assumed to be ^A(ctrl-a). Please refer to this link for more details regarding hive ddl operations. Q3: create table mydb.user (uid int,name string) row format delimited fields terminated by '\t' ; If we define table with specified delimiter so that’s the reason why you are able to see delimiter as \t in desc formatted table command. ... View more

@Chris Grren We cannot add Stored State values to the flow files attributes, How ever you can create custom processor which can implement the logic of getting current batch timestamp attribute added to each flowfile. You can also use RestApi call to the AttributeRollingWindow processor to know what is the current batch timestamp value. We can use Get method with make a call to the below path to get the current state of the processor GET /processors/{id}/state //Gets the state for a processor For more details about Rest Api calls refer to this link. Example:- As i'm having AttributeRollingWindow processor with processor id as d372cfe5-0162-1000-398e-62f2d3652166 and from command line we need to make rest api call as below. Making Rest Api call from Command Line:- $ curl -i -X GET -H 'Content-Type:application/json' http://localhost:8080/nifi-api/processors/d372cfe5-0162-1000-398e-62f2d3652166/state Response:- as you can view below these are the states saved in the processor and the first key is our current batch timestamp i.e 1524017336077 { "componentState" : { "componentId" : "d372cfe5-0162-1000-398e-62f2d3652166", "stateDescription" : "Store the values backing the rolling window. This includes storing the individual values and their time-stamps or the batches of values and their counts.", "localState" : { "scope" : "LOCAL", "totalEntryCount" : 7, "state" : [ { "key" : "1524017336077", //current batch timestamp "value" : "1.0" }, { "key" : "1524017336079", "value" : "1.0" }, { "key" : "1524017336082", "value" : "1.0" }, { "key" : "1524017336083", "value" : "1.0" }, { "key" : "1524017336085", "value" : "1.0" }, { "key" : "1524017336086", "value" : "1.0" }, { "key" : "count", "value" : "7" } ] } } } States stored in the processor:- Making same Rest Api call from NiFi processor:- use InvokeHttp processor with Remote URL as http://localhost:8080/nifi-api/processors/d372cfe5-0162-1000-398e-62f2d3652166/state Output:- From the response relationship you will get same response as above { "componentState" : { "componentId" : "d372cfe5-0162-1000-398e-62f2d3652166", "stateDescription" : "Store the values backing the rolling window. This includes storing the individual values and their time-stamps or the batches of values and their counts.", "localState" : { "scope" : "LOCAL", "totalEntryCount" : 7, "state" : [ { "key" : "1524017336077", //current batch timestamp "value" : "1.0" }, { "key" : "1524017336079", "value" : "1.0" }, { "key" : "1524017336082", "value" : "1.0" }, { "key" : "1524017336083", "value" : "1.0" }, { "key" : "1524017336085", "value" : "1.0" }, { "key" : "1524017336086", "value" : "1.0" }, { "key" : "count", "value" : "7" } ] } } } As the first state array value will be your current batch timestamp. . If the Answer Addressed your question, Click on Accept button below to accept the answer, That would be great help to Community users to find solution quickly for these kind of issues. ... View more