Member since
08-01-2021
59
Posts
14
Kudos Received
7
Solutions
My Accepted Solutions
| Title | Views | Posted |
|---|---|---|
| 3896 | 11-18-2022 09:06 AM | |
| 6385 | 11-15-2022 05:46 PM | |
| 4091 | 10-12-2022 03:18 AM | |
| 3068 | 10-11-2022 08:52 AM | |
| 7458 | 10-08-2022 08:23 AM |
09-20-2026
10:23 PM
Greetings dear Apache NiFi community. I've recently had the opportunity to use NiFi to read and write JSON data to PostgresSQL using its native JSONB data type. While researching how to do this, I found very limited information, mainly just one helpful StackOverflow question and reply by @mpayne. My use case was fairly simple: I had JSON files that I wanted to store as-is and be able to retrieve them later in the exact same form. This presented three main challenges:
JSONB is a column, meaning that when writing to it, you need to wrap up your JSON under a new field with a name matching the column.
PutDatabaseRecord will not work if the data under the new wrapper field is a valid JSON, since it will try to parse it as record values. Therefore, you need to escape your JSON data before writing to the DB.
When reading the data back, you need to restore the escaped JSONs back to their original form and remove the wrapper field from step 1.
Writing to JSONB column
In my case, the PostgreSQL table had two columns:
data of type JSONB
id of type UUID (generated from the contents of data)
The table DDL is as such:
CREATE TABLE IF NOT EXISTS <schema>.<table> (
data JSONB NOT NULL,
id UUID GENERATED ALWAYS AS ((data->>'id')::uuid) STORED,
PRIMARY KEY (id)
)
To wrap my JSONs in the 'data' field and escape them, I used a single ReplaceText processor configured as such:
Replacement Strategy
Regex Replace
Search Value
(?s)(^.*$)
Replacement Value
{"data": "${'$1':escapeJson():replace('\\', '\\\\')}"}
Character Set
UTF-8
Maximum Buffer Size
1 MB
Evaluation Mode
Entire text
Line-by-Line Evaluation Mode
All
Credit to Mark Payne (@mpayne) for suggesting that replacement value in the StackOverflow thread mentioned above. The expression matches your entire file content (careful if your JSONs are bigger than the Maximum Buffer Size), making it accessible via the '$1' attribute in the Replacement Value property. This allows you to run Expression Language on your entire file's contents. We use this to run :escapeJson():replace('\\', '\\\\') on the JSON data to prepare it for insertion. There is also a simple {"data": "...."} literal text wrapper so that the following PutDatabaseRecord processor will know to write this escaped json to the 'data' column.
No special configuration is needed in the PutDatabaseRecord processor. This is the overall flow:
Note that although it appears you insert escaped JSONs to your DB, they are actually stored correctly and can be queried using standard JSONB operators (->, ->>, etc.)
Reading the Data back
To read the data back, I use ExecuteSQLRecord to fetch rows (making sure to specify Array Output in the JSON RecordWriter), followed by another ReplaceText processor to edit the results back to their correct shape.
I define a basic SQL query to fetch just the data column:
SELECT data FROM <schema>.<table>
The resulting FlowFile content might look something like:
[
{
"data": "{\"id\": \"488b5b63-6ccf-4816-9a38-38c3befd248f\", \"myKey\": \"myValue\", \"keyTwo\": \"bababa\"}"
},
{
"data": "{\"id\": \"2fec9cad-5246-4d92-aa41-9a57cfc4e8ab\", \"myKey\": \"myValue23\", \"keyTwo\": \"21312\"}"
},
{
"data": "{\"id\": \"6efbac98-3a10-422d-a2a2-ac6d79ffef6b\", \"myKey\": \"myValue23\", \"keyTwo\": {\"nested\": [\"abc\", \"def\"]}}"
}
]
As you can see, each row gets read as a flat JSON where the key is 'data' like the column name, and the value is a stringified/escaped version of the original content.
To fix it now, I use this config in the following replace text:
Replacement Strategy
Regex Replace
Search Value
"data"\s*:\s*"\{((?:\\.|[^"\\])*)\}"
Replacement Value
${'$1':unescapeJson()}
Character Set
UTF-8
Maximum Buffer Size
1 MB
Evaluation Mode
Line-By-Line
Line-by-Line Evaluation Mode
All
Note that I use Line-By-Line mode this time, since ExecuteSQLRecord seems to fetch rows prettified with newlines separating them (Even if Pretty Print is set to false in the RecordWriter!). Because it scans line-by-line, it ignores the braces/brackets at the top and bottom of the response and in-between the different rows, and you can instead focus on just the content of the rows, which looks like:
"data" : "{\"id\": \"488b5b63-6ccf-4816-9a38-38c3befd248f\", \"myKey\": \"myValue\", \"keyTwo\": \"bababa\"}"
Thus, we match everything within the value of data's braces "data" : "{ <this part> }" and once again use ReplaceText's special expression syntax '$1' to refer to our escaped version of the original JSON content and then run :unescapeJson() on it to restore it back to its valid form.
The output will look like
[ {
"id" : "488b5b63-6ccf-4816-9a38-38c3befd248f",
"myKey" : "myValue",
"keyTwo" : "bababa"
}, {
"id" : "2fec9cad-5246-4d92-aa41-9a57cfc4e8ab",
"myKey" : "myValue23",
"keyTwo" : "21312"
}, {
"id" : "6efbac98-3a10-422d-a2a2-ac6d79ffef6b",
"myKey" : "myValue23",
"keyTwo" : {
"nested" : [ "abc", "def" ]
}
} ]
And now you have valid JSONs to work with again for the rest of your flows. DISCLAIMER: An external user contributed to this article. Cloudera may not verify that the steps may be applicable for all use cases and may be very specific to a particular distribution. Please follow with caution and at your own risk. If needed, raise a support case to get confirmation.
... View more
Labels:
01-27-2026
12:46 PM
1 Kudo
@Green_ The MapCacheClientService does not provide any TTL capability. It simply communicates with the MapCacheServer. The MapCacheServer has some eviction strategies available, but none are based on some TTL attribute. The RedisDistributedMapCacheClientService controller service can be configured with a TTL; however, it can not be set dynamically via a FlowFile Attribute since it does not support NiFi Expression Language. This would require an improvement to this controller service. I'd recommend creating a Apache NiFi Jira (https://issues.apache.org/jira/projects/NIFI/issues) requesting such an improvement with your use case. Ideally this would involve enabling support for NiFi Expression language so that each FlowFile could pass a unique TTL. Would also need to handle scenario when a FlowFile is missing the TTL attribute: - Use some default TTL. - Have separate "default TTL" configurable property that is used when Attribute is not set on FlowFile. - Have FlowFile route to failure when attribute not set (this might not be possible). Please help our community grow. If you found any of the suggestions/solutions provided helped you with solving your issue or answering your question, please take a moment to login and click "Accept as Solution" on one or more of them that helped. Thank you, Matt
... View more
01-14-2026
06:30 AM
1 Kudo
@MattWho Wow! I think this pattern would work best for my usecase. I hadn't even considered the first challenge you brought up of production flows having their Parameter Context unassigned if I were to update their version. That would've been painful to find out after deploying many instances. Back in NiFi 1 I used to handle situations such as this with variables, since they could just be directly attached to Process Groups and so I never had to worry about creating separate objects (parameters) and ensuring they get attached, or that every new instance of a versioned flow had to have its own unique context created. It's been a couple years but I believe I even questioned Pierre about this in one of his appearances in the Israeli NiFi meet-ups. In regards to product work, I've ran into this case of trying to use NiFi as the underlying tool for different SaaS platforms multiple times already. There could definitely be some QoL changes made to make such a use-case easier to implement with NiFi's flow registry, I guess the responsibility lies in people like me opening issues to bring them though 🙂 Thank you very much for the suggestions Matt! Green
... View more
01-06-2026
05:25 AM
@MuruganFinastra Since you are getting a 403 response, the first thing you should do is see what user identity this 403 is being returned for. For this you'll want to be tailing the nifi-user.log while you attempt to make this rest-api call. You will see the denied related log lines in the nifi-user.log. That logging will provide the user identity string and which NiFi authorization policy required for which that user identity did not have the required permissions. Using this output, we can determine the next steps required here. Is the expected user identity being logged? What is the logged authorization policy resulting in the 403 response? Also which user authentication and authorization configuration options are you using in your setup? Please help our community grow. If you found any of the suggestions/solutions provided helped you with solving your issue or answering your question, please take a moment to login and click "Accept as Solution" on one or more of them that helped. Thank you, Matt
... View more
09-20-2025
09:15 AM
If "Infer Schema" isn't working this is likely a bug. Could you provide an example JSON and the error message that happens during schema inference?
... View more
07-31-2024
11:38 AM
@Green_ I had the same issue can you please share the detailed steps of how you resolved it? As I'm new to NiFi.
... View more
09-21-2023
12:42 AM
I am also experiencing this issue when attempting to write data to a redis in cluser mode. Did you find a solution or workaround @sofronic ?
... View more
12-12-2022
05:49 AM
Hello Eyal, thanks once again for your interest in helping out. I have not expressed myself correctly. I believe there is no backpressure, as I confirmed visually on the nifi UI. On the upper banner I see there are globally about 5000 flowfiles circulating at a particular time. I see some of the queues have 20, 50, 100, 200 flowfiles waiting at an instant. The queue that has more queued flow files is the kafka producer processor, that has about 2000. I believe these quantities are not sufficient to trigger backpressure-caused adjustment of scheduling of tasks. Apache Benchmark sends concurrent http requests. In this case, i set each of the 5 client instances to send each 1000 requests concurrently. I have thread pool sized to 96 and assigned 16 concurrent tasks to all of the critical path processors. Despite this, I get a low core load average (between 2 and 7), the disk's IOPS are in the dozens per second (the maximum that the disks allow is 3000). I'm all out of ideas at this point. 😄 Thanks for the support!
... View more
12-09-2022
02:00 PM
@F_Amini @Green_ Is absolutely correct here. You should be careful when increasing concurrent tasks as just blindly increasing it everywhere can have the opposite effect on throughput. I recommend stopping setting the concurrent tasks back to 1 or maybe 2 on all the processor where you have adjusted away from the default of 1 concurrent task. Then take a look at the processor further downstream in your dataflow where it has a red input connection but black (no backpressure) outbound connections. This processor s @Green_ mentioned is the processor causing all your upstream backlog. Your'll want to monitor your CPU usage as you make small incremental adjustments to this processors concurrent tasks until you see the upstream backlog start to come down. If while monitoring CPU, you see it spike pretty consistently at 100% usage across all your cores, then your dataflow has pretty much reached the max throughput it can handle for yoru specific dataflow design. At this point you need to look at other options like setting up a NiFi cluster where this work load can be spread across multiple servers or designing your datafow differently with different processors to accomplish same use case that may have a lesser impact on CPU (not always a possibility). Thanks, Matt
... View more
12-01-2022
02:33 AM
Hi, thanks for the details. Unfortunately it is not working. I get an empty array [] as output. I have tried it with extract and split mode. I applied the schema text property as suggested with "NestedKey" and "nestedValue" as name. None gives me an output. Meanwhile I have achieved what I wanted using SplitContent and then again another jolt processor. Of course it would be more elegant if I could make it work with ForkRecord.
... View more