@Abdelkrim Hadjidj Today I have a solution for just my case: Because the key of our data is unique string on json data. So here my step to capture both key and value of json data: Step 1: Using ExtractText to obtain key (project_1, project_2,...) Step 2: Using ReplaceText to replace key on json data by a static key (project_1, project_2,... string will be replaced by content string. Step 3: Using EvaluateJsonPath to obtain data of content above and transform it to flowfile-content for next processor. Step 4: Using PutMongoDB to insert data into corresponding database using Expression Regex {projectname} with data above. As alternatively way, we also can use ExecuteScript to convert key and value of json data to attribute then insert into MongoDB. But as I running, the throughput when use the first way is higher then using ExecuteScript. Here my code when using ExecuteScript: #!/usr/bin/python3
import json
import java.io
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback
class StreamCallback(StreamCallback):
def __init__(self):
pass
def process(self, inputStream, outputStream):
text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
data = json.loads(text)
for key in data:
first = key
break
content = data[first]
contentString = json.dumps(content)
viewFlowFile = session.create()
viewFlowFile = session.putAllAttributes(viewFlowFile,
{'project': str(first), 'content': contentString})
session.transfer(viewFlowFile, REL_SUCCESS)
flowFile = session.get()
if flowFile != None:
flowFile = session.write(flowFile, StreamCallback())
session.transfer(flowFile, REL_FAILURE)
session.commit()
... View more
