I have a sequence of flowfiles that I need to put to mongo. A sample is at the bottom of this question. The flowfile contains JSON with a field "evt_data": {"stdds_id": "stdds_id_value" .. I need that stdds_id_value to be the key for update/upsert into mongo. I'm looking for help with the flow configuration that will make this work. My current PutMongo (non-working attempt) configuration is this: I'm trying to put a record that is keyed on the evt_data.stdds_id value as see in the following flowfile JSON: In other words the key for the document in Mongo would be "TARGET-STDDS-KCLT-157481920517" This doesnt work, and I end up seeing multiple documents in mongo for that same key. I should only see one per distinct key. What is the proper way to set the update key? ... View more

I need to do an update/upsert into Mongo. Essentially the command that I need to run is seen in the following command (this works in the mongo command line client): Notice in the first update I search for the document that matches the specified STDDS_DST_ID. In the second update, I match any of several Ids including the one that was already matched. In this simple example. I have a set of linked Ids: TFM_ID, FVIEW_ID, STDDS_DST_ID. The set of linked IDs is unique. For in this example that per-set distinction guarantees that you wont find STDDS_DIST_ID 100 associated with another FVIEW_ID, or TFM_ID. You'll only find it with FVIEW_ID 3000 and TFM_ID 300000. So assuming that I have a flowfile that contains some number of fields (e.g. fld1, fld2, fl3), and one or more of the ids: TFM_ID, STDDS_DST_ID, FVIEW_ID, how can I configure PutMongo so that it will update/upsert the appropriate document (that one that matches one or more of these IDs)? Again, in my mind, PutMongo simply needs to be configured consistent with the update you see in the image above. I just dont have much experience with PutMongo. Looking at the documentation, I believe I must do the following: Set Mode to update Set Upsert to true Leave Update Query Key Empty Set the Update Query to something such as the first argument in the sample command: { $or: [{"TFM_ID": "300000"}, {"FVIEW_ID": "3000"},{"STDDS_DST_ID": "100"}]} Set the flowfile content to the data to place in the document (including the $set): {$set: {"fld2": "fld2_val", "fld3": "fld3_val", "TFM_ID": "300000", "FVIEW_ID": "3000","STDDS_DST_ID": "100"}} Are my assumptions on the mark? ... View more

I have a flowfile that contains a list of JSON objects. For each JSON object in the list, I need to generate a single call to PutMongo to place that object into Mongo. The key will be the evt_data.targetDataId. The call to PutMongo will insert or update the record for that targetDataId. I'm looking for the most efficient way to process such an array. Do I need to first need to use an intermediate processor (e.g. SplitJSON) to reduce it to a list of flowfiles, one per object in the array, before sending to PutMongo? If so, what processor should I use for that? For example, if I have a flowfile that looks like the following. This would result in 2 inserts (or updates) using PutMongo. It would be really nice if PutMongo supported this conversion of an array of documents, to an array of "inserted/updated" documents, out of the box. [ { "evt_type":"wheels_down", "callsign":"C887", "evt_data":{ "targetDataId":633901919919, "sourceName":"STDDS", "callsign":"C887", "sourceSpecificGufi":[ "KORD_C887_2715", "KORD_5868622" ], "updateTime":1563478106000, "targetType":"VEHICLE", "subSource":"KORD" } }, { "evt_type":"wheels_up", "callsign":"C888", "evt_data":{ "targetDataId":633901919919, "sourceName":"STDDS", "callsign":"C888", "sourceSpecificGufi":[ "KORD_C888_2715", "KORD_5868622" ], "updateTime":1563478106000, "targetType":"VEHICLE", "subSource":"KORD" } } ] ... View more

I am looking to create a customNIFI processor. It receives flowfiles, but in order to generate the resulting flowfile, it must first reach out to a backend service to get information that must be added to the output flowfile. I am currently thinking of accessing the backend service in a worker thread in the custom processor. Once the additional information is received, I'd then queue this for an asynchronous creation of an output flowfile (in onTrigger). Is this a reasonable thing to do in a custom processor? Typically (using the egress pattern - or many of the other common patterns- documented on the NIFI developer page), input flowfiles are immediately (synchronously) transformed into an output flowfile. However that is not the pattern I have in mind.. Input flowfiles will instead be asynchrnously transformed into output flowfiles (and this might not even be a 1 to 1 mapping from input flowfile to output flowfile). In this regard the processor will have behaviour similar to MergeContent, or MergeRecord (which are indeed asynchronously generating output flowfiles). ... View more

I am looking to create a customNIFI ingest processor. It does not receive flowfiles, but it does ingest data from a number of backend services and then it produces flowfiles. The process of accumulating data from the backend services happens over a timeline, so my thought is that I should likely create a worker thread that executes this data accumulation/ingestion timeline, and then places the accumulated data into a threadsafe queue to then be handled onTrigger. The OnTrigger method would then pull data from this queue, and create the flowfiles to be produced. Is this a reasonable approach? Are there alternative approaches that the NIFI experts might recommend? ... View more

Hi. Can someone please point me to an example of how to use the nifi PutMongo processor to update one or several documents in Mongo? Even better would be an example of how to update a document if it matches some flowfile criteria, and to add a document to mongo if that criteria is not matched. I'll update this question in just a minute with an example of the mongo document in question. ... View more

According to the instructions found here: https://nifi.apache.org/quickstart.html Instructions in essence: git clone https://gitbox.apache.org/repos/asf/nifi.git git checkout master mvn install -Dmaven.test.skip=true I've built nifi successfully in the past. I'm trying to clone and build the current master. I see the following failure: Why is the current nifi not building? ... View more

I've used the PutHDFS processor as I've started to understand how to deal with big data environments. Up until now I've been putting very small files into HDFS. This seems to be architecturally bad practice. The HDFS block size defaults to about 128 MB, and the hadoop community recommendation seems to be that applications (that write to HDFS) should write files that are GB in size, or even TB. I'm trying to understand how to do this with Nifi. Part of my concern is a concern for the data analysts. What is the best way to logically structure files that are appropriate for HDFS? Currently the files that I am writing contain small JSON objects, or lists. I use MergeRecord to intentionally make the file I write larger. However my JSON objects accumulate fast thousands of JSON records per second potentially. For the Big Data/Nifi experts, I'd appreciate any thoughts relative to the best way to use Nifi to support streaming large data objects into HDFS. ... View more

I'm trying to do a PutHDFS to a directory on a computer running HADOOP in single node fashion. I've verified that hadoop is properly setup. I have also verified the connection from the Nifi PutHDFS processor to hadoop. However I see a permission denied. I'd like to solve this the "right way" using kerberos. I'm looking for the cleanest tutorial focused on this. I've found a bunch of fragments on the web, but nothing that is clean and complete. ... View more

We are building a custom processor, that receives network messages in a binary format. Currently those messages (flowfiles) are coming routed to our processor from a ListenUDP processor. We are decoding that format, and converting to a JSON format. It looks like we need to deal with an endian byte swap. What is the cleanest way for us to implement this byte swap? Presumably a small piece of code in "OnTrigger" will do the job. I am hoping to find an example NAR that already does this. The best solution would be one where we detect the need for a byte swap and then implement it if needed. In this fashion we will ensure that the custom processor works on little endian or big endian architectures. Currently we are running on an Intel x86 architecture (little endian). ... View more

My simple flow: The input XML: The output JSON: ConvertRecord Configuration: XMLReader Configuration: Several questions arise: Why are position and ncsmTrackData null? Why are all the attributes on fltdMessage being ignored? How can I see the inferred schema? What is a proper schema to correctly transform those two null fields, (along with the rest of the transformation as it currently looks ok)? ... View more

This seems like a bug. The exception I see is as follows: 2019-04-12 09:49:26,810 ERROR [Timer-Driven Process Thread-6] o.a.n.processors.standard.ConvertRecord ConvertRecord[id=016a1079-25cf-16fd-fbfa-785d65603736] Failed to write MapRecord[{dpTransitionFix=MTU, star=MapRecord[{routeType=STAR, routeName=VKTRY2}], departureFixAndTime=MapRecord[{fixName=MTU, arrTime=2019-04-12T14:06:55Z}], starTransitionFix=MDANO, rvsmData=MapRecord[{futureCompliance=true, equipped=true, currentCompliance=true}], routeOfFlight=KSLC.RUGGD1.MTU..HBU..PNH..MDANO.VKTRY2.KDFW/1551, dp=MapRecord[{routeType=DIRECT, routeName=RUGGD1}], diversionIndicator=NO_DIVERSION, eta=MapRecord[{etaType=ESTIMATED, timeValue=2019-04-12T15:50:43Z}], etd=MapRecord[{timeValue=2019-04-12T13:47:00Z, etdType=ESTIMATED}], nextPosition=MapRecord[{longitudeDecimal=-111.9777733, latitudeDecimal=40.7883933}], flightTraversalData2=MapRecord[{waypoint=[Ljava.lang.Object;@6aca8e9d, fix=[Ljava.lang.Object;@35df6693, airway=[Ljava.lang.Object;@4300900c, center=[Ljava.lang.Object;@550b388b, sector=[Ljava.lang.Object;@6b0bfad1}], arrivalFixAndTime=MapRecord[{fixName=VKTRY, arrTime=2019-04-12T15:40:09Z}]}] with schema [] as a JSON Object due to java.lang.ClassCastException: java.lang.ClassCastExceptionjava.lang.ClassCastException: null 2019-04-12 09:49:26,810 ERROR [Timer-Driven Process Thread-6] o.a.n.processors.standard.ConvertRecord ConvertRecord[id=016a1079-25cf-16fd-fbfa-785d65603736] Failed to write MapRecord[{dpTransitionFix=MTU, star=MapRecord[{routeType=STAR, routeName=VKTRY2}], departureFixAndTime=MapRecord[{fixName=MTU, arrTime=2019-04-12T14:06:55Z}], starTransitionFix=MDANO, rvsmData=MapRecord[{futureCompliance=true, equipped=true, currentCompliance=true}], routeOfFlight=KSLC.RUGGD1.MTU..HBU..PNH..MDANO.VKTRY2.KDFW/1551, dp=MapRecord[{routeType=DIRECT, routeName=RUGGD1}], diversionIndicator=NO_DIVERSION, eta=MapRecord[{etaType=ESTIMATED, timeValue=2019-04-12T15:50:43Z}], etd=MapRecord[{timeValue=2019-04-12T13:47:00Z, etdType=ESTIMATED}], nextPosition=MapRecord[{longitudeDecimal=-111.9777733, latitudeDecimal=40.7883933}], flightTraversalData2=MapRecord[{waypoint=[Ljava.lang.Object;@6aca8e9d, fix=[Ljava.lang.Object;@35df6693, airway=[Ljava.lang.Object;@4300900c, center=[Ljava.lang.Object;@550b388b, sector=[Ljava.lang.Object;@6b0bfad1}], arrivalFixAndTime=MapRecord[{fixName=VKTRY, arrTime=2019-04-12T15:40:09Z}]}] with schema [] as a JSON Object due to java.lang.ClassCastException: java.lang.ClassCastExceptionjava.lang.ClassCastException: null The data that I am sending into convertrecord typically is processed fine. Sometimes it fails to process a record as seen here (notice 7 failures and many successes): Is the best action to report this on the NIFI bug tracker? ... View more

The data I have is as follows: In the above case $[0].fltdOutput.fltdMessage is an array, shown in its original form. I will see this in two forms: object and array. In the following case $[0].fltdOutput.fltdMessage is an object. [ {   "fltdOutput" : {     "fltdMessage" : {       "ncsmFlightTimes" : {         "qualifiedAircraftId" : {           "aircraftId" : "DAL2160",           "computerId" : {             "facilityIdentifier" : "TFMS",             "idNumber" : 927           },           "gufi" : "KF45658600",           "igtd" : "2019-04-12T13:30:00Z",           "departurePoint" : {             "airport" : "KDFW"           },           "arrivalPoint" : {             "airport" : "KSLC"           }         },         "flightStatusAndSpec" : {           "flightStatus" : "FILED",           "aircraftModel" : "A319",           "aircraftspecification" : "A319"         },         "etd" : null,         "eta" : null,         "ctd" : "1970-01-01T00:00:00Z",         "cta" : "1970-01-01T00:00:00Z",         "rvsmData" : null,         "departureFixAndTime" : null,         "arrivalFixAndTime" : null       }     }   } } ] Sometimes $[0].fltdOutput.fltdMessage is an array: [ {   "fltdOutput" : {     "fltdMessage" : [ {       "trackInformation" : {         "qualifiedAircraftId" : {           "aircraftId" : "SWA2148",           "computerId" : {             "facilityIdentifier" : "KZDC",             "idNumber" : 507           },           "gufi" : "KJ42198600",           "igtd" : "2019-04-12T12:35:00Z",           "departurePoint" : {             "airport" : "KMCO"           },           "arrivalPoint" : {             "airport" : "KROC"           }         },         "speed" : 476,         "reportedAltitude" : {           "assignedAltitude" : {             "simpleAltitude" : "390C"           }         },         "position" : {           "latitude" : {             "latitudeDMS" : "MapRecord[{seconds=31, minutes=04, degrees=34, direction=NORTH}]"           },           "longitude" : {             "longitudeDMS" : "MapRecord[{seconds=09, minutes=51, degrees=080, direction=WEST}]"           }         },         "timeAtPosition" : "2019-04-12T13:48:58Z",         "ncsmTrackData" : {           "eta" : "MapRecord[{etaType=ESTIMATED, timeValue=2019-04-12T15:05:36Z}]",           "rvsmData" : "MapRecord[{futureCompliance=true, equipped=true, currentCompliance=true}]",           "departureFixAndTime" : "MapRecord[{fixName=WORMS, arrTime=2019-04-12T13:15:00Z}]",           "nextEvent" : "MapRecord[{longitudeDecimal=-80.8368083, latitudeDecimal=34.1616639}]"         }       },       "flightPlanAmendmentInformation" : null,       "arrivalInformation" : null,       "flightPlanInformation" : null     }, {       "trackInformation" : {         "qualifiedAircraftId" : {           "aircraftId" : "UAL817",           "computerId" : {             "facilityIdentifier" : "KZDC",             "idNumber" : "18T"           },           "gufi" : "KH29739600",           "igtd" : "2019-04-12T11:00:00Z",           "departurePoint" : {             "airport" : "KAUS"           },           "arrivalPoint" : {             "airport" : "KEWR"           }         },         "speed" : 486,         "reportedAltitude" : {           "assignedAltitude" : {             "simpleAltitude" : "410C"           }         },         "position" : {           "latitude" : {             "latitudeDMS" : "MapRecord[{seconds=54, minutes=47, degrees=37, direction=NORTH}]"           },           "longitude" : {             "longitudeDMS" : "MapRecord[{seconds=05, minutes=51, degrees=080, direction=WEST}]"           }         },         "timeAtPosition" : "2019-04-12T13:48:58Z",         "ncsmTrackData" : {           "eta" : "MapRecord[{etaType=ESTIMATED, timeValue=2019-04-12T14:46:21Z}]",           "rvsmData" : "MapRecord[{futureCompliance=true, equipped=true, currentCompliance=true}]",           "arrivalFixAndTime" : "MapRecord[{fixName=ARD, arrTime=2019-04-12T14:31:57Z}]",           "departureFixAndTime" : "MapRecord[{fixName=ILEXY, arrTime=2019-04-12T12:03:06Z}]",           "nextEvent" : "MapRecord[{longitudeDecimal=-80.76862798233492, latitudeDecimal=37.80398852749686}]"         }       },       "flightPlanAmendmentInformation" : null,       "arrivalInformation" : null,       "flightPlanInformation" : null     } ]   } } ] What is the RouteOnContent configuration that will allow me to determine if $[0].fltdOutput.fltdMessage is an array? I'm trying to setup a test flow that only performs SplitJSON if it is an array. Notice in the following, I have a failure out of SplitJSON because its handling a version of the file that isnt an array I believe that I have to use a regular expression (similar to ^\[\{ ) as suggested here. However I'm not sure how the regular expression would look for $[0].fltdOutput.fltdMessage I think the following should work I've tested this regular expression online: ^\[{"fltdOutput":{"fltdMessage":\[ Yet NIFI complains The following shows that the regular expression is valid, and that there are two matches within the five test strings. ... View more