Nifi – When Record-Oriented Doesn't Pay Off
{
"type": "record",
"name": "Mdm_Record",
"namespace": "any.org",
"fields": [
{
"name": "metadata",
"type": {
"type": "record",
"name": "MessageInfo",
"namespace": "any.org",
"fields": [
{
"name": "ts",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
}
}
]
}
},
{
"name": "data",
"type": {
"type": "record",
"name": "MDM_Item",
"fields": [
{
"name": "object",
"type": {
"type": "record",
"name": "Node",
"fields": [
{
"name": "guid",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "template",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "language",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "fields",
"type": {
"type": "map",
"values": [
"Node",
{
"type": "record",
"name": "Barcode",
"fields": [
{
"name": "provider",
"type": {
"type": "record",
"name": "Provider",
"fields": [
{
"name": "name",
"type": "string"
},
{
"name": "city",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "taxIdNumber",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "taxRegistrationReasonCode",
"type": [
"null",
"string"
],
"default": null
}
]
}
},
{
"name": "barcode",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "supplierArticle",
"type": [
"null",
"string"
],
"default": null
}
]
},
{
"type": "record",
"name": "MediaResource",
"fields": [
{
"name": "link",
"type": "string",
},
{
"name": "filename",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "mimetype",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "fileSize",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "format",
"type": [
"null",
"string"
],
"default": null
}
]
},
{
"type": "record",
"name": "SofaSpecification",
"fields": [
{
"name": "seats",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "bedLength",
"type": {
"type": "record",
"name": "QuantityValue",
"fields": [
{
"name": "value",
"type": "float",
},
{
"name": "measureUnit",
"type": [
"null",
{
"type": "record",
"name": "MeasureUnit",
"fields": [
{
"name": "unitName",
"type": "string",
},
{
"name": "unitCode",
"type": "string",
}
]
}
],
"default": null
}
]
}
},
{
"name": "bedWidth",
"type": "QuantityValue",
},
{
"name": "backHeightFolded",
"type": "QuantityValue",
},
{
"name": "backHeightFoldedOut",
"type": "QuantityValue",
},
{
"name": "fullLength",
"type": "QuantityValue",
}
]
},
{
"type": "record",
"name": "MattressSpecification",
"fields": [
{
"name": "bedLength",
"type": "QuantityValue",
},
{
"name": "bedWidth",
"type": "QuantityValue",
},
{
"name": "height",
"type": "QuantityValue",
},
{
"name": "maximumBedLoad",
"type": "QuantityValue",
}
]
},
"QuantityValue",
"MeasureUnit",
{
"type": "array",
"items": [
"Node",
"Barcode",
{
"type": "record",
"name": "ItemPack",
"fields": [
{
"name": "packageName",
"type": "string",
},
{
"name": "package",
"type": [
"null",
{
"type": "record",
"name": "Pack",
"fields": [
{
"name": "packageGuid",
"type": "string",
},
{
"name": "packageType",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "packaging",
"type": {
"type": "map",
"values": [
{
"type": "record",
"name": "Box",
"fields": [
{
"name": "length",
"type": "QuantityValue",
},
{
"name": "width",
"type": "QuantityValue",
},
{
"name": "height",
"type": "QuantityValue",
}
]
},
{
"type": "record",
"name": "SoftPackaging",
"fields": [
{
"name": "twist",
"type": "boolean",
}
]
}
]
},
}
]
}
],
"default": null
},
{
"name": "length",
"type": "QuantityValue",
},
{
"name": "width",
"type": "QuantityValue",
},
{
"name": "height",
"type": "QuantityValue",
},
{
"name": "volume",
"type": "QuantityValue",
},
{
"name": "weightNetto",
"type": "QuantityValue",
},
{
"name": "weightGross",
"type": "QuantityValue",
},
{
"name": "seatsCount",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "packagedUnitsCount",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "setSeatsCount",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "priority",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "measureUnit",
"type": [
"null",
"MeasureUnit"
],
"default": null
},
{
"name": "mainLogisticsPackaging",
"type": [
"null",
"boolean"
],
"default": null
},
{
"name": "barcode",
"type": [
"null",
"string"
],
"default": null
}
]
},
"MediaResource",
"string",
"int",
"boolean"
]
},
{
"type": "map",
"values": [
"string",
"int",
"boolean"
]
},
"string",
"int",
"boolean"
]
}
}
]
}
},
{
"name": "stateName",
"type": "string"
}
]
}
}
]
}
Since the previously chosen option with the Jolt transformation did not work, I decided to use a proven means for extreme cases – a script. There is a good processor – ScriptedTransformRecord, which allows you to process one record at a time, getting an object of the type Record.
The script turned out to be quite large due to the presence of map types, as well as the possibility of the “null” value in the required fields. Since there is a possibility of modifying the data format or adjusting S2T, it will be necessary to change the script in the future, which, given its large size, entails an increase in complexity for the developer. So I decided to abandon the script and return to the developed and debugged specification, because for JSON it works correctly, the error occurs only when working with a record, when the data scheme is applied.
That is, in this case, I decided to abandon the recording processing and move on to processing the entire content, which leads to an increase in the number of FlowFiles and, accordingly, an increase in the use of RAM.
In NiFi, you can apply the Jolt specification to records using JoltTransformRecord or use the JoltTransformJSON processor, which expects JSON as input and transforms it not as a record, but as a single JSON file. Since the entire content is processed, it is better to input a single JSON object rather than an array to reduce overhead. Therefore, it was necessary to first split the incoming FlowFile into fragments, where each would contain one JSON. This allowed the transformation for a single object to be performed quickly, but entailed the generation of a large number of FlowFiles. For this, I used SplitRecord, where Reader read the Avro format, and RecordSetWriter was configured to write a single JSON object.