@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

@Arash ,   It's likely that your flow is running on multiple nodes and each node is trying to process every file. When different nodes try to write the same file at the same time you get these errors.   You can change your flow to use a List-Fetch pattern. The ListHDFS processor should run only on the primary node and the output of it (the list of files) can be load-balanced across the nodes so that each node will FetchHDFS different files.   Cheers, André   ... View more

@totokogure ,   The EvaluateJsonPath that you added is extracting the $.hits array and storing it as an attribute. The next processor (SplitJson), does not even use that attribute, though. It extracts $.hits again from the flowfile content. The EvaluateJsonPath in this flow should be unnecessary. If you connected InvokeHTTP to SplitJson, things should work correctly.   Cheers, André ... View more