I am dealing with Kafka dataset where there are multiple types of message data is processing (coming) Sample data: eventType 1- { "type": "record", "name": "Dispatch_Accepted", "namespace": "accepted.avro", "fields": [ { "name": "John", "type": "string", "doc": "Name of the user account" }, { "name": "email", "type": "string", "doc": "The email of the user logging message on the blog" }, { "name": "timestamp", "type": "long", "doc": "time in seconds" } ], "doc:": "A basic schema of Dispatch_Rejected" } EventType-2 { "type": "record", "name": "Dispatch_Rejected", "namespace": "rejected.avro", "fields": [ { "name": "Merry", "type": "string", "doc": "Name of the user" }, { "name": "email", "type": "string", "doc": "The email of the user logging message on the blog" }, { "name": "timestamp", "type": "long", "doc": "time in seconds" } ], "doc:": "A basic schema Rejected data" } Schema of the data getting validated from Confluent Schema Regisry (Working Fine), I need to apply filter on Schema name (Dispatch_Rejected and Dispatch_Accepted) and crete two separate data files for each so I am using QueryRecord Processor which below query <Dispatch_Rejected>=Select * from FLOWFILE WHERE name='Dispatch_Rejected' <Dispatch_Accepted>=Select * from FLOWFILE WHERE name='Dispatch_Accepted' This is not working.. can't identify the schema name. Controller service is working fine. 1- How I can pick the schema name from Controller service 2- Should I need to assign the value ${schema.name} in another variable <My_schema> and need to write SELECT Statement like <Dispatch_Rejected>=Select * from FLOWFILE WHERE My_Schema.name='Dispatch_Rejected' <Dispatch_Accepted>=Select * from FLOWFILE WHERE My_Schema.name='Dispatch_Accepted' Summary-- I want to filter the data based on eventType, and create separate data files Please help
... View more
