Hello I'm Samuel and I work on Miguel's team,
there is a single point of the flow where a queue is filling, and backpressure is being applied: a logattribute to register the ips of the incoming requests:
Apart from that, there are roughly 5000 flowfiles flowing in the system at a given moment when were processing a steady flow of events, which given the fact that end-to-end latency is approx 1second, seems acceptable.
The scenario in which we are obtaining these results is a development environment, where we are performing stress tests using apache benchmark. So the production rate is steady.
The requests are directed at a load balancer. The failed requests seem to originate from that load balancer, which is filling its queue, and dropping incoming requests when that happens.
1.UpdateRecord
I could not find a freeform text processor. Here is an example of the incoming data in the form of query parameters:
TS=20221108-093700&UID=12345&ETID=22&EGID=44&DID=Hello%20World!&HHID=492489&OID=666
What reader could be used to transform free text to json?
2. transforming null values:
an example of the incoming data:
{
"TS": "20221108-093700",
"UID": 12345,
"ETID": "22",
"EGID": 44,
"DID": "Hello World!",
"HHID": 0,
"OID": 666
}
when event type id is 11, and the householdid is either 0,"" or " ", we want to change the HHID to null
I could not find a way to change the value to null and cannot evaluate one field and transform another. Is it possible to transform a value to null (the value null and not the string "null")? What is being done wrong in the example below?
[
{
"operation": "shift",
"spec": {
"TS": "TimeStamp",
"UID": "UserID",
"OID": "OpCoID",
"ETID": "EventTypeID",
"EGID": "EventGroupID",
"DID": "DeviceID",
"HHID": {
"0": {
"#null": "HouseholdID"
},
"*": {
"@(2,HID)": "HouseholdID"
}
},
"*": "Custom.&"
}
},
{
"operation": "default",
"spec": {
"Custom": {}
}
}
]
3. UpdateAttribute + RouteContent -> QueryRecord
We're doing control flow: evaluating if a given field is numeric, then we're seeing whether there was any error, and based on the messageError attribute I'm routing to success and error. With the query we can apply a transformation but cannot decide based on the evaluation of the sql query. Is this interpretation correct in your opinion?
Greetings
there is a single point of the flow where a queue is filling, and backpressure is being applied: a logattribute to register the ips of the incoming requestsHave you tried increasing the "Run Duration" property in the scheduling tab of the LogAttribute processor?
Sometimes setting this even at 25ms can considerably increase your flow's throughput. I'd recommend playing around with this setting and seeing what works best.
1.UpdateRecord
I could not find a freeform text processor. Here is an example of the incoming data in the form of query parameters:
TS=20221108-093700&UID=12345&ETID=22&EGID=44&DID=Hello%20World!&HHID=492489&OID=666
What reader could be used to transform free text to json?
If you claim to receive the data in the form of query parameters, perhaps you can use one of the following. For starters, query parameters should be added to your received flowfiles as attributes IF you use the HandleHttpRequest and HandleHttpResponse processors (which ideally you should be doing instead of using the limited ListenHTTP processor). All query parameters received by the HandleHttpRequest processor are added as attributes named http.query.param.<parameter name>. If it is only the query parameters you wish to process in your flow, you can now use the AttributesToJSON processor and list all these query param attributes to generate a json more strictly. If you want to process the query params in addition to your request body, I'd like you to give me a more detailed example of input/output and I'll post a better solution.
2. I didn't 100% understand your use-case. Regarding inserting null values (and not the string 'null'), I found this thread in which a solution is offered. Essentially, you first use your shift operation to get rid of keys for which you want to insert null, then you use a default operation and add null as the default value. I created the following spec and it seems to inject null correctly:
[
{
"operation":"shift",
"spec":{
"TS":"TimeStamp",
"UID":"UserID",
"OID":"OpCoID",
"ETID":"EventTypeID",
"EGID":"EventGroupID",
"DID":"DeviceID",
"HHID":{
"0":{
"null":"HouseholdID"
},
" ":{
"null":"HouseholdID"
},
"":{
"null":"HouseholdID"
},
"*":{
"@(2,HID)":"HouseholdID"
}
},
"*":"Custom.&"
}
},
{
"operation":"default",
"spec":{
"Custom":{
},
"HouseholdID": null
}
}
]If you only wish to inject nulls when ETID == 11, the transformation will be a bit more complex.. I think it would be easier as a whole to just extract ETID and HHID to attributes and then use expression language with the "ifElse" operation to evaluate what will be shifted into HouseholdId.
3. I'm afraid I didn't completely understand this point either. If the issue is routing to specific relationships with QueryRecord, every dynamic property you add will create a new relationship. In order for a flowfile to pass to the right relationship, you should add a WHERE clause to your sql query that will only match flowfiles that should pass to that specific route/relationship.
I'd be glad to help more 🙂! If you still have questions, providing examples (expected input/output) will make it easier for me to offer a solution for the different parts in your flow.
My regards,
Eyal.
Hello Eyal.,
thank you for taking the time to respond, it has been helpful in learning more about what works in nifi processors.
Increasing the Run Duration of the log processor eliminated the queued flowfiles and that was good.
The baseline throughput of the system is about 5600 req/s before any change.
The installation includes 6 t5.4xlarge ec2 instances (each has 16 threads)
I have been trying to change the flow but also trying to tune the nifi configurations. About the changes to the flow:
1. Parsing query parameters
I used the http.query.param.YYY attributes with the AttributesToJson processor instead of parsing the http query string. This replaced 6 processors by 1 and the performance gain was about 5% (registered aprox 5800 req/s)
2.
When http.query.param.ETID = 11 and http.query.param.HHID is in (0,'0','') i affected http.query.param.HHID with a special value that it should not have (it is supposed to contain either integers of numeric characters), and then removed http.query.param.HHID.
Then I applied the jolt expression that you shared. Here, because the approach is probably not the best, there was no significant performance gain.
3.
I used the queryrecord processor to test if a field is numeric with an sql query, and in that case routed it to the error processing
I used a select * from Flowfile with a where clause contrary to the first to route it to non-error processing.
Here, the performance gain was also about 5%.
So up to this point I there were about 6,1k events/second, no major breakthrough acchieved that can be attributed to changes in the nifi flow.
I consider this far from what i expected.
As to the nifi configurations tuning:
So the first thing i did was changing the provenance repository to in-memory and that had a good impact on the performance. A gain of about 600 req/s.
At this point I had 3-node cluster and added 3 more nodes. The first tests that i did i was well impressed, i got consistently above 7700 req/s. At one point I got about 14200 req/s.
I could never reproduce the results again, as subsequent tests registered a decrease in performance.
I saw some posts that seemed to describe a situation similar to mine. A cluster that had good performance, then got worse results. The tickets indicated that there could be a garbage collector interference, and reported that the initial throughput was acchieved again after restart of the service. This did not happen in my case. The performance degraded.
I tried to change some of the configurations in the nifi.properties safety valve following this post:
https://community.cloudera.com/t5/Community-Articles/HDF-CFM-NIFI-Best-practices-for-setting-up-a-hi...
I changed all the properties to the indicated values and got no performance increase.
In summary, the system was performing at 5600 req/s with 3 nodes. After adding 3 nodes it got above 7700, sometimes above 9000 and once at 14200 req/s. Now it is performing at 6600 req/s with 6 nodes.
I get the idea that the degradation in performance per node is not to be attributed to the changes i did to the flow, but do not have sufficient information to identify a cause.
Greetings,
Samuel
