https://community.cloudera.com/t5/Support-Questions/Apache-Nifi-QueryRecord-based-on-FlowFile-attribute/m-p/310093#M224022 <P><SPAN>I'm using&nbsp;</SPAN><STRONG>CaptureChangeMySQL</STRONG><SPAN>&nbsp;Processor to capture MySQL event data. for the&nbsp;</SPAN>update<SPAN>&nbsp;event I got the following data in my FlowFile.</SPAN></P><P>&nbsp;</P><PRE>{ "type" : "<SPAN class="hljs-keyword">update</SPAN><SPAN class="hljs-string">", "</SPAN><SPAN class="hljs-built_in">timestamp</SPAN><SPAN class="hljs-string">" : 1610120620000, "</SPAN>binlog_filename<SPAN class="hljs-string">" : "</SPAN>delta<SPAN class="hljs-number">.000002</SPAN><SPAN class="hljs-string">", "</SPAN>binlog_position<SPAN class="hljs-string">" : 1943, "</SPAN><SPAN class="hljs-keyword">database</SPAN><SPAN class="hljs-string">" : "</SPAN><SPAN class="hljs-keyword">test</SPAN><SPAN class="hljs-string">", "</SPAN>table_name<SPAN class="hljs-string">" : "</SPAN>Employee<SPAN class="hljs-string">", "</SPAN>table_id<SPAN class="hljs-string">" : 92, "</SPAN><SPAN class="hljs-keyword">columns</SPAN><SPAN class="hljs-string">" : [ { "</SPAN><SPAN class="hljs-keyword">id</SPAN><SPAN class="hljs-string">" : 1, "</SPAN><SPAN class="hljs-keyword">name</SPAN><SPAN class="hljs-string">" : "</SPAN><SPAN class="hljs-keyword">id</SPAN><SPAN class="hljs-string">", "</SPAN>column_type<SPAN class="hljs-string">" : 4, "</SPAN><SPAN class="hljs-keyword">last_value</SPAN><SPAN class="hljs-string">" : 1, "</SPAN><SPAN class="hljs-keyword">value</SPAN><SPAN class="hljs-string">" : 1 }, { "</SPAN><SPAN class="hljs-keyword">id</SPAN><SPAN class="hljs-string">" : 2, "</SPAN><SPAN class="hljs-keyword">name</SPAN><SPAN class="hljs-string">" : "</SPAN><SPAN class="hljs-keyword">Name</SPAN><SPAN class="hljs-string">", "</SPAN>column_type<SPAN class="hljs-string">" : 12, "</SPAN><SPAN class="hljs-keyword">last_value</SPAN><SPAN class="hljs-string">" : "</SPAN>Khalid<SPAN class="hljs-string">", "</SPAN><SPAN class="hljs-keyword">value</SPAN><SPAN class="hljs-string">" : "</SPAN>John<SPAN class="hljs-string">" }, { "</SPAN><SPAN class="hljs-keyword">id</SPAN><SPAN class="hljs-string">" : 3, "</SPAN><SPAN class="hljs-keyword">name</SPAN><SPAN class="hljs-string">" : "</SPAN>Email<SPAN class="hljs-string">", "</SPAN>column_type<SPAN class="hljs-string">" : 12, "</SPAN><SPAN class="hljs-keyword">last_value</SPAN><SPAN class="hljs-string">" : "</SPAN>lacus.varius@utsem.net<SPAN class="hljs-string">", "</SPAN><SPAN class="hljs-keyword">value</SPAN><SPAN class="hljs-string">" : "</SPAN>lacus.varius@utsem.net<SPAN class="hljs-string">" }, { "</SPAN><SPAN class="hljs-keyword">id</SPAN><SPAN class="hljs-string">" : 4, "</SPAN><SPAN class="hljs-keyword">name</SPAN><SPAN class="hljs-string">" : "</SPAN>City<SPAN class="hljs-string">", "</SPAN>column_type<SPAN class="hljs-string">" : 12, "</SPAN><SPAN class="hljs-keyword">last_value</SPAN><SPAN class="hljs-string">" : "</SPAN>Luziânia<SPAN class="hljs-string">", "</SPAN><SPAN class="hljs-keyword">value</SPAN><SPAN class="hljs-string">" : "</SPAN>Luziânia<SPAN class="hljs-string">" } ] }</SPAN><BR /><BR /></PRE><P>then I'm using<SPAN>&nbsp;</SPAN><STRONG>EvaluateJsonPath</STRONG><SPAN>&nbsp;</SPAN>Processor to add attributes like<SPAN>&nbsp;</SPAN><STRONG>event_type,timestamp,table_name,value of column name id</STRONG></P><P>then I want to use the<SPAN>&nbsp;</SPAN><STRONG>RouteOnAttribute</STRONG><SPAN>&nbsp;</SPAN>processor to query other tables apart from the set attribute name (table_name) to merge all records in a single JSON that has the same id.</P><HR /><P>Now the main problem is that if more tables updated at the same time it will generate extra record(which is completely useless also it will query other tables multiple times)</P><P>So, I decided to use<SPAN>&nbsp;</SPAN><STRONG>QueryRecord</STRONG><SPAN>&nbsp;</SPAN>Processor and Set<SPAN>&nbsp;</SPAN><STRONG>Run Schedule</STRONG><SPAN>&nbsp;</SPAN>at 2-5 minutes.</P><P>Now I want to Query FlowFiles like</P><LI-CODE lang="markup">SELECT * FROM FLOWFILES WHERE '$.timestamp' = (SELECT MAX('$.timestamp') FROM FLOWFILE) GROUP BY '$.id'</LI-CODE><P><STRONG>It means It will fetch those flow files which has a unique id and Maximum timestamp value.</STRONG><SPAN>&nbsp;</SPAN>this query is not working. so how can I use attributes in the<SPAN>&nbsp;</SPAN><STRONG>QueryRecord</STRONG><SPAN>&nbsp;</SPAN>Processor to achieve this scenario? or is there any better way to handle this?</P> Wed, 20 Jan 2021 10:54:44 GMT Achyut 2021-01-20T10:54:44Z