You can use PutMongoRecord for this, the JsonTreeReader can accept "one JSON per line" (as of NiFi 1.7.0 via NIFI-4456, also HDP 3.2) ... View more

session.get() only returns a FlowFile reference (or null if one is not available). Instead you want something like session.get(9) which will return a List<FlowFile> guaranteed not to be null, but it may have size < 9 so your check should work in this case. Having said that, are you sure you want to transfer them to failure? If you're planning on just routing them back to the ExecuteScript (until 9 are available) then you could just do session.rollback() or session.transfer(flowFile, Relationship.SELF) ... View more

The ReplaceText puts the CLOB data directly into the content, which is why the error message occurs. You could use a PreparedStatement to get around this, but you very likely don't want to use an attribute as the CLOB is large. The alternative is PutDatabaseRecord, you can provide a JsonTreeRecordReader (with schema) and it will generate the prepared statement but use the content to get the CLOB data rather than an attribute. If they won't know the schema (but it only contains primitive types) then they could use InferAvroSchema if necessary. However it is preferable to provide the schema directly via the Record Reader and/or a Schema Registry. ... View more

You should be able to use ValidateRecord for this, although your reader schema might need to treat the incoming field as an int/long so it will read in negative numbers successfully. Then your write schema can have that field with its correct type (date, e.g.) You can also use QueryRecord for this, it allows you to query the flow file fields using SQL, such as "SELECT * FROM FLOWFILE WHERE columnDate >= 0" or whatever is appropriate. ... View more

Can you explain more about your use case? Why do you need to extract ~100 fields, how are they being used downstream? I wonder if a record-based processor or something would be more appropriate... ... View more

If you know the fields, you can use JoltTransformJSON on the original input JSON so you don't have to use SplitJson, here's a spec that will do the (explicit) field name conversion: [ { "operation": "shift", "spec": { "data": { "Table": { "*": { "Age": "data.Table[#2].age", "FirstName": "data.Table[#2].first_name", "LastName": "data.Table[#2].last_name" } } } } } ] You could also use UpdateRecord, you'd just need separate schemas for the JsonTreeReader and JsonRecordSetWriter. ... View more

You can use JoltTransformJSON to convert this JSON into just the "name: value" pairs used by PutDatabaseRecord, here is a spec that will do it: [{ "operation": "shift", "spec": { "columns": { "*": { "@(value)": "[#1].@(1,name)" } } } }] If you find that PutDatabaseRecord is slow, it's likely because you're putting one record at a time into the DB. Instead consider using MergeRecord to bundle together more CDC events, then you can use the following spec to transform all of them for use in PutDatabaseRecord as a "micro-batch": [{ "operation": "shift", "spec": { "*": { "columns": { "*": { "@(value)": "[#4].@(1,name)" } } } } }] ... View more

Maybe try without the string() function around it?! I'm not sure, since I used the transform above and it worked... ... View more