Hi,   Not sure what is the problem with the current situation. Usually when the first process group finishes and send the flowfile to the second process group, this flowfile will be dropped when executing the the "GenerateTableFetch" in the second group and new flowfiles will be generated based on that. Another option if you dont want to have any relationships between the first and second process groups through output\input ports and make them completely independent where no flowfiles are transferred between them is to use the Nifi Rest Api to trigger the second group GenerateTableFetch from the first group using the InvokeHttp Processor. Hope that helps. Thanks ... View more

Hi, You dont need to have programming language to learn Expression Language. It can help but its not required. ... View more

  Hi,   Try the following:   [ { "operation": "shift", "spec": { "*": { "*": "[#2].&", "$": "[#2].product" } } } ]   If that helps please accept solution. Thanks ... View more

Hi, I dont think you catch the SQL error in the sense that PutSQL wont report the error. However you can use the PutDatabaseRecrod instead and use the failure relationship to LogMessage where you can access the error message using the " putdatabaserecord.error " attribute. A better way of capturing errors from any processor (Global Catch) is to use the SiteToSiteBulletinReportingTask as explained here: SiteToSiteBulletinReportingTask   If that helps please accept solution. Thanks   ... View more

Hi, Not sure if you are looking for the exact thing but this should give you the expected output from the sample you provided:   [ // Flatten an array of photo objects into a prefixed // soup of properties. { "operation": "shift", "spec": { "content": { "*": { "*": { "*": { "$": "error", "$1": "product", "$2": "ErrorType", "@": "q" } } } } } } ] If that helps, please accept solution. Thanks ... View more

@Bello as this is an older post, you would have a better chance of receiving a resolution by starting a new thread. This will also be an opportunity to provide details specific to your environment that could aid others in assisting you with a more accurate answer to your question. You can link this thread as a reference in your new post. ... View more

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

Hi, thanks for the details. Unfortunately it is not working. I get an empty array [] as output. I have tried it with extract and split mode. I applied the schema text property as suggested with "NestedKey" and "nestedValue" as name. None gives me an output. Meanwhile I have achieved what I wanted using SplitContent and then again another jolt processor. Of course it would be more elegant if I could make it work with ForkRecord. ... View more

@SAMSAL Thank you for your help.  ... View more

@Mohamed_Shaaban  I recommend starting a new community question with the details specific to your setup.  This allows the community to address/assist with your specific setup versus comparing your issue to what was shared in this post. Thanks, Matt ... View more

I think everything works fine, the json flowfiles are the response of an invokeHttp request to an API, I receive more than 5000 flowfiles, each flowfile contains 200 records. Problems happen only with 4 files that contain the ETX symbol. ... View more

Hi, I'm only able to do that via two processors: 1- JotlTransfomJSON : this is to add desired attributes (attr_id & attr_name) to the flowfile json using the following spec:   [ { "operation": "shift", "spec": { "#${attr_id}":"attr_id", "#${attr_name}":"attr_name", "model": "model", "cars_drivers": { "*": { "id": "cars_drivers[#2].id", "name": "cars_drivers[#2].name", "is_processed": "cars_drivers[#2].is_processed" } } } } ]   2 - UpdateRecord Processor: Once the attributes are added to the Json , you can update those records Is_processed value only when the id matches the attr_id and name matches the attr_name. To do that set the following properties:      a- Replacement Value Strategy: Literal Value      b- /cars_drivers[*]/id[.=../../../attr_id]/../name[.=../../../attr_name]/../is_processed : 1   Hope that helps, if it does please accept solution.   ... View more

Thanks for the reply, it worked 🙂   Also, I got the expected output and didn't see the `MapRecord` as displayed in your output   Thanks! ... View more

Hello bro @SAMSAL , were you successfully control the nifi component using postman? If yes, would it be okay to share your countermeasure. Currently encountering the same issue that you had. Thanks in advance.   ... View more

How did anyone solve this issue? I was currently trying to solve similar issue via nifi api (Postman) ... View more

Input 1 "npi_code": "NEW", "cifa_category": "XM", "o9_category": "Accessories", "equipment_type": "AC", "equipment_sub_type": "CASEPROT", "model_manf": "137181-1050", "model_comcast": "Samsung Galaxy A51 Presidio Grip", "memory": "", "color": "Black", "os": "", "upc": 849000000000, "sku": 849000000000, "cifa_id": 273271, "manufacturer_part": "137181-1050", "description": "Speck Samsung Galaxy A51 Presidio Grip - Black", "uom": "", "manufacturer": "SPK", "serialized": "N", "equip_id_type": "ICCID", "pre_inserted_sim": "", "sim_type": "", "sim_form": "", "base_feature_tier": "", "market_attractiveness": "", "memory_tier": "", "size_tier": "", "planning_item": 848709087461, "stat_item": "Samsung Galaxy A51 Presidio Grip", "model_mem": "Samsung Galaxy A51 Presidio Grip", "model_mem_color": "Samsung Galaxy A51 Presidio Grip", "hero_status": "", "launch_date": "2020-05-08", "dummy_item_creation_date": "", "npi_announcement_date": "", "pre_order_date": "", "ga_date": "", "landed_at_launch_date": "", "planned_end_of_purchase": "", "planned_end_of_replen": "", "planned_end_of_forecast": "", "independence_date": "", "disable_date": "2021-04-29", "accessory_retail_end_of_life": "" }   Input 2 { "Item": [ { "npi_code": "NEW", "cifa_category": "XM", "o9_category": "Accessories", "equipment_type": "AC", "equipment_sub_type": "CASEPROT", "model_manf": "137181-1050", "model_comcast": "Samsung Galaxy A51 Presidio Grip", "memory": "", "color": "Black", "os": "", "upc": 849000000000, "sku": 849000000000, "cifa_id": 273271, "manufacturer_part": "137181-1050", "description": "Speck Samsung Galaxy A51 Presidio Grip - Black", "uom": "", "manufacturer": "SPK", "serialized": "N", "equip_id_type": "ICCID", "pre_inserted_sim": "", "sim_type": "", "sim_form": "", "base_feature_tier": "", "market_attractiveness": "", "memory_tier": "", "size_tier": "", "planning_item": 848709087461, "stat_item": "Samsung Galaxy A51 Presidio Grip", "model_mem": "Samsung Galaxy A51 Presidio Grip", "model_mem_color": "Samsung Galaxy A51 Presidio Grip", "hero_status": "", "launch_date": "2020-05-08", "dummy_item_creation_date": "", "npi_announcement_date": "", "pre_order_date": "", "ga_date": "", "landed_at_launch_date": "", "planned_end_of_purchase": "", "planned_end_of_replen": "", "planned_end_of_forecast": "", "independence_date": "", "disable_date": "2021-04-29", "accessory_retail_end_of_life": "" }, { "npi_code": "NEW", "cifa_category": "XM", "o9_category": "Accessories", "equipment_type": "AC", "equipment_sub_type": "CASEPROT", "model_manf": "137181-1050", "model_comcast": "Samsung Galaxy A51 Presidio Grip", "memory": "", "color": "Black", "os": "", "upc": 849000000000, "sku": 849000000000, "cifa_id": 273271, "manufacturer_part": "137181-1050", "description": "Speck Samsung Galaxy A51 Presidio Grip - Black", "uom": "", "manufacturer": "SPK", "serialized": "N", "equip_id_type": "ICCID", "pre_inserted_sim": "", "sim_type": "", "sim_form": "", "base_feature_tier": "", "market_attractiveness": "", "memory_tier": "", "size_tier": "", "planning_item": 848709087461, "stat_item": "Samsung Galaxy A51 Presidio Grip", "model_mem": "Samsung Galaxy A51 Presidio Grip", "model_mem_color": "Samsung Galaxy A51 Presidio Grip", "hero_status": "", "launch_date": "2020-05-08", "dummy_item_creation_date": "", "npi_announcement_date": "", "pre_order_date": "", "ga_date": "", "landed_at_launch_date": "", "planned_end_of_purchase": "", "planned_end_of_replen": "", "planned_end_of_forecast": "", "independence_date": "", "disable_date": "2021-04-29", "accessory_retail_end_of_life": "" } ] ... View more

Thanks @SAMSAL, I was tracking that previous post but since it was so old I was hoping newer versions of NiFi could address this.  This is such a simple thing with XML and the EvaluateXPath processor, I am just surprised that we can't do something similar with JSON.  Oh well, 2 processors it is. ... View more

Hi, Try the following Jolt spec:   [ { "operation": "shift", "spec": { "tags": { "*": { "tag": "tags.[#2].Parameter", "value": "tags.[#2].value" } } } } ] If you find this helpful please accept solution. ... View more

There should be two controllers configured:   DistributedMapCacheClientService DistributedMapCacheServer Client is, eh, client, server is a backend for your cache. Without server client has nowhere to send a flowfile. ... View more

Thank you all. I eventually evaluated the json path to extract the url. My mind was astray as i was using complex solution to a simple problem.  ... View more

Brilliant!  Exactly what I was looking for.  Although it seems a little peculiar to me that we need to rely on a Jolt transform for this operation and not the UpdateRecord processor. Particularly since NiFi makes it a point to discuss Arrays and Maps in the documentation. Thanks for the Jolt transform because I spent a lot of time trying to get the Jolt transform to work and couldn't quite figure it out.  Now I see what I was doing wrong.   ... View more

Hello,   you can configure that just 1 Flowfile will be handled inside of a specific ProcessorGroup.  There is the config option on PG which is called: Process Group FlowFile Concurrency for that you can set the value: Single FlowFile per Node   After PutKudu you will destroy the FlowFile or route it out from ProcessorGroup, then the next FlowFile will be released to enter to ProcessorGroup   In your case the Flow would look like: ListFile Processor -> ProcessorGroup (handels fetchFile, data Transformation and putKudu) ... View more

Hi, Please try the following spec: [ { "operation": "shift", "spec": { "timestamp": { "*": { "@(2,resourceid)": "[&1].resourceid", "@": "[&1].timestamp" } }, "Key": { "*": { "@": "[&1].key" } }, "data": { "*": { "@": "[&1].data" } } } } ] If you find this helpful, please accept solution. Thanks ... View more