@belka ,   No, Apache Druid is not currently supported.   Cheers, André   ... View more

@Omarb ,   Initially I thought this was a problem with the CSVRecordSetWriter, but I was mistaken. The issue here is that even though your CSVReader is set to ignore the header line, it has Schema Access Strategy set to "Infer Schema", and this will cause the reader to consume the first line of the flow file to infer the schema, even though the other property tells it to ignore it.   To avoid this, set the Schema Access Strategy property to "Use 'Schema Text' Property" and provide a schema that matches your flowfile structure. For example: "type": "record", "name": "MyFlowFile", "fields": [ { "name": "col_a", "type": "string" }, { "name": "col_b", "type": "string" }, { "name": "col_c", "type": "string" }, ... ] }   This will stop the first line being "consumed" by the reader.   Cheers, André ... View more

@Griggsy ,   I don't know if there's a way to do exactly that with either JOLT or ReplaceText processors. You can do it with ExecuteScript and the following Python script, though: from org.apache.commons.io import IOUtils from java.nio.charset import StandardCharsets from org.apache.nifi.processor.io import StreamCallback import json import re def to_snake_case(name): return re.sub(r'(?<!^)(?=[A-Z])', '_', name).lower() def convert_keys_to_snake_case(obj): if isinstance(obj, list): return [convert_keys_to_snake_case(o) for o in obj] else: return {to_snake_case(k): v for k, v in obj.items()} class PyStreamCallback(StreamCallback): def __init__(self): pass def process(self, inputStream, outputStream): text = IOUtils.toString(inputStream, StandardCharsets.UTF_8) converted = convert_keys_to_snake_case(json.loads(text)) outputStream.write(bytearray(json.dumps(converted).encode('utf-8','ignore'))) flow_file = session.get() if flow_file != None: flow_file = session.write(flow_file, PyStreamCallback()) session.transfer(flow_file, REL_SUCCESS)   Cheers, André ... View more

@nk20 ,   You can use the NiFi API to extract the current state of any stateful processor if you need this. See the example below for a PutFile processor: access_token=$(curl -k \ -X POST \ -H 'Content-Type: application/x-www-form-urlencoded' \ -d 'username=admin&password=supersecret1' "https://localhost:8443/nifi-api/access/token") curl -k \ -H "Authorization: Bearer $access_token" \ "https://localhost:8443/nifi-api/processors/dd24aaec-0182-1000-ffff-ffff9f128d94/state" { "componentState": { "componentId": "dd24aaec-0182-1000-ffff-ffff9f128d94", "stateDescription": "After performing a listing of files, the timestamp of the newest file is stored. This allows the Processor to list only files that have been added or modified after this date the next time that the Processor is run. Whether the state is stored with a Local or Cluster scope depends on the value of the <Input Directory Location> property.", "clusterState": { "scope": "CLUSTER", "totalEntryCount": 0, "state": [] }, "localState": { "scope": "LOCAL", "totalEntryCount": 6, "state": [ { "key": "id.0", "value": "/tmp/hsperfdata_nifi/129", "clusterNodeId": "420e540a-ccd6-4e7c-bc25-c572f503b338", "clusterNodeAddress": "nifi1:8443" }, { "key": "id.0", "value": "/tmp/hsperfdata_nifi/130", "clusterNodeId": "00e23669-130f-4e12-8a26-be3ab95923d4", "clusterNodeAddress": "nifi0:8443" }, { "key": "listing.timestamp", "value": "1661567544049", "clusterNodeId": "420e540a-ccd6-4e7c-bc25-c572f503b338", "clusterNodeAddress": "nifi1:8443" }, { "key": "listing.timestamp", "value": "1661567541525", "clusterNodeId": "00e23669-130f-4e12-8a26-be3ab95923d4", "clusterNodeAddress": "nifi0:8443" }, { "key": "processed.timestamp", "value": "1661567544049", "clusterNodeId": "420e540a-ccd6-4e7c-bc25-c572f503b338", "clusterNodeAddress": "nifi1:8443" }, { "key": "processed.timestamp", "value": "1661567541525", "clusterNodeId": "00e23669-130f-4e12-8a26-be3ab95923d4", "clusterNodeAddress": "nifi0:8443" } ] } } }   Cheers, André ... View more