@yagoaparecidoti ,   The opposite command doesn't exist, as far as I know. Querying the database is the only option that I'm aware.   Cheers, André   ... View more

@sathish3389 ,   One way to do this is to use the ExecuteScript processor with a script like the one below. This script with set the attribute dv_sys_id with the content that you want and will also add that to the content of the flowfile. from org.apache.commons.io import IOUtils from java.nio.charset import StandardCharsets from org.apache.nifi.processor.io import StreamCallback import json class PyStreamCallback(StreamCallback): def __init__(self): self.dv_sys_id = None def process(self, inputStream, outputStream): text = IOUtils.toString(inputStream, StandardCharsets.UTF_8) obj = json.loads(text) self.dv_sys_id = 'dv_sys_id=%s' % (','.join(obj['dv_sys_id'],)) outputStream.write(bytearray(self.dv_sys_id.encode('utf-8','ignore'))) flow_file = session.get() if flow_file != None: callback = PyStreamCallback() flow_file = session.write(flow_file, callback) flow_file = session.putAttribute(flow_file, 'dv_sys_id', callback.dv_sys_id) session.transfer(flow_file, REL_SUCCESS)     Cheers, André   ... View more

@lie ,   MySQL 8 is not a supported version for CM/CDH 6.3.1, which is the version you're using, as you can see here: https://docs.cloudera.com/documentation/enterprise/6/release-notes/topics/rg_database_requirements.html   You should pick a version that is compatible with your CDH version. Alternatively, you can upgrade to CDP.   Cheers, André ... View more

@Curry5103 ,   ListenSyslog can receive a very large amount of data and it may be hard to match the throughput of that with an InvokeHTTP processor.   To avoid dropping syslog messages, especially if using UDP protocol, you will probably need to provide enough buffer for those messages to decouple the receipt of the messages from the InvokeHTTP execution. You can do that in NiFi by increase the maximum queue size limits, or maybe use something like Kafka, where you can temporarily store the syslog messages and another flow can read from Kafka and call InvokeHTTP for each of them.   Cheers, André   ... View more

@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

@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