Thank you very much!! ... View more

Hi @SAMSAL ,   Ugh!  That is a little disconcerting to hear about the "wrapper" and "insertion" strategies.  I resorted to using the "wrapper" strategy and then follow up with a Jolt transform to get the data into the format I want.  However, if you think using SQL is a more bullet proof option then I'm open to that if you know how to write a SQL that uses RecordPath to put the data into the structure I need.     Here is an example dataset, which gives me the exact same outcome.  The incoming flow file is OpenSky flight data, and the primary key is "hex".  Using an UpdateRecord processor, I add the field  "aircraft": "placeholder" because the JoinEnrichment processor requires that the field already exist before inserting the enrichment data.   The second dataset is my enrichment data, which is a record in a large dataset called UnitedAirlinesFleet, and "icao24" is the foreign key.  So "hex" and "icao24" are the two keys. I want to do something very simple, which is to insert the enrichment data into flight data as a record where the field name is "aircraft", as shown in the third dataset below.  I believe this is precisely what the insertion strategy is designed to do.    I really appreciate you SAMSAL and your expertise.  Your suggestion to use the ForkEnrichment and JoinEnrichment was really helpful.  I subsequently found a Cloudera video on this topic and that also helped a lot with how to structure the flow.  Thank you.       { "hex" : "a39fb1", "flight" : "UAL798 ", "alt_baro" : 23000, "alt_geom" : 23200, "gs" : 415.2, "track" : 275.9, "baro_rate" : 2432, "squawk" : "1467", "emergency" : "none", "category" : "A3", "nav_qnh" : 1013.6, "nav_altitude_mcp" : 26016, "nav_heading" : 270.0, "lat" : 39.969681, "lon" : -105.610657, "nic" : 8, "rc" : 186, "seen_pos" : 1.1, "version" : 2, "nic_baro" : 1, "nac_p" : 9, "nac_v" : 1, "sil" : 3, "sil_type" : "perhour", "gva" : 2, "sda" : 2, "mlat" : [ ], "tisb" : [ ], "messages" : 876, "seen" : 1.1, "rssi" : -1.8, "aircraft" : "placeholder" }           { "icao24" : "a39fb1", "registration" : "N33203", "manufacturericao" : "BOEING", "manufacturername" : "Boeing", "model" : "737-824", "typecode" : "B738", "serialnumber" : 30613, "linenumber" : null, "icaoaircrafttype" : "L2J", "operator" : null, "operatorcallsign" : "UNITED", "operatoricao" : "UAL", "operatoriata" : null, "owner" : "Wells Fargo Trust Co Na Trustee", "testreg" : null, "registered" : null, "reguntil" : "2023-07-31", "status" : null, "built" : "2000-01-01", "firstflightdate" : null, "seatconfiguration" : null, "engines" : "CFM INTL. CFM56 SERIES", "modes" : false, "adsb" : false, "acars" : false, "notes" : null, "categoryDescription" : "No ADS-B Emitter Category Information" }           { "hex" : "a39fb1", "flight" : "UAL798 ", "alt_baro" : 23000, "alt_geom" : 23200, "gs" : 415.2, "track" : 275.9, "baro_rate" : 2432, "squawk" : "1467", "emergency" : "none", "category" : "A3", "nav_qnh" : 1013.6, "nav_altitude_mcp" : 26016, "nav_heading" : 270.0, "lat" : 39.969681, "lon" : -105.610657, "nic" : 8, "rc" : 186, "seen_pos" : 1.1, "version" : 2, "nic_baro" : 1, "nac_p" : 9, "nac_v" : 1, "sil" : 3, "sil_type" : "perhour", "gva" : 2, "sda" : 2, "mlat" : [ ], "tisb" : [ ], "messages" : 876, "seen" : 1.1, "rssi" : -1.8, "aircraft" : { "icao24" : "a39fb1", "registration" : "N33203", "manufacturericao" : "BOEING", "manufacturername" : "Boeing", "model" : "737-824", "typecode" : "B738", "serialnumber" : 30613, "linenumber" : null, "icaoaircrafttype" : "L2J", "operator" : null, "operatorcallsign" : "UNITED", "operatoricao" : "UAL", "operatoriata" : null, "owner" : "Wells Fargo Trust Co Na Trustee", "testreg" : null, "registered" : null, "reguntil" : "2023-07-31", "status" : null, "built" : "2000-01-01", "firstflightdate" : null, "seatconfiguration" : null, "engines" : "CFM INTL. CFM56 SERIES", "modes" : false, "adsb" : false, "acars" : false, "notes" : null, "categoryDescription" : "No ADS-B Emitter Category Information" } }       ... View more

That worked perfectly. Thank you.  ... View more

Hi, You can do that in two ways: 1- Calling P2 from P1 using Output\Input Ports to connect process groups 2- Using Nifi API using nifi HttpInvoke Processor from P1 to control the run-state of the Initial Processor in P2. Hope that help. Thanks ... View more

I had a similar issue, it happens because of nifi not being able to resolve the schema. You need to first create a schema and set the type of billing address as string and add it to avroschemaregistry and lastly configure your jsontreereader to use that schema ... View more

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