You can use QueryRecord for this. Ensure your JSONReader's schema has the geaendertAm_ABAS field marked as a timestamp type (not a string), such as: {
"namespace": "nifi",
"name": "ABAS",
"type": "record",
"fields": [
{"name": "ID","type": "int"},
{"name": "geaendertAm_ABAS","type": {"type": "long","logicalType": "timestamp-millis"}}
]
} Then you can add a user-defined property (let's call it "max") to QueryRecord with the value SELECT MAX(geaendertAm_ABAS) from FLOWFILE Your JSONRecordSetWriter will need a schema with just the field: {
"namespace": "nifi",
"name": "ABAS",
"type": "record",
"fields": [
{"name": "geaendertAm_ABAS","type": {"type": "long","logicalType": "timestamp-millis"}}
]
} Once you click the Apply button on QueryRecord, you will be able to create a connection from QueryRecord called "max" and connect it to the next downstream processor. As an alternative, here is a Groovy script for use in an ExecuteScript processor, note that it is very specific to your input: def flowFile = session.get()
if(!flowFile) return
try {
flowFile = session.write(flowFile, {inputStream, outputStream ->
def objList = new groovy.json.JsonSlurper().parse(inputStream)
def max = objList.max {Date.parse("yyyy-MM-dd'T'HH:mm:ss.SSSSSSS'Z'",it.geaendertAm_ABAS)}
def maxOutput = "{\"geaendertAm_ABAS\": \"${max.geaendertAm_ABAS}\"}"
outputStream.write(maxOutput.bytes)
} as StreamCallback)
session.transfer(flowFile, REL_SUCCESS)
} catch(e) {
log.error("Error while determining max", e)
session.transfer(flowFile, REL_FAILURE)
} If you instead want the max in an attribute, you can use something like: def flowFile = session.get()
if(!flowFile) return
try {
def inputStream = session.read(flowFile)
def objList = new groovy.json.JsonSlurper().parse(inputStream)
def max = objList.max {Date.parse("yyyy-MM-dd'T'HH:mm:ss.SSSSSSS'Z'",it.geaendertAm_ABAS)}
inputStream.close()
flowFile = session.putAttribute(flowFile, 'MAX_geaendertAm_ABAS', max.geaendertAm_ABAS.toString())
session.transfer(flowFile, REL_SUCCESS)
} catch(e) {
log.error("Error while determining max", e)
session.transfer(flowFile, REL_FAILURE)
}
... View more
