Append Value to cache with FetchDistributedMapCache / PutDistributedMapCache
À users
Hello !
I am new to Nifi and I'm trying to find the best workflow to update or
append content to the MapCache
I created my DistributedMapCacheServer and Service.
I have several simple flowlogs with Content Like this:
{
"eventType" : "api",
"message" : "blabla"
}
and simple Attribute "ServiceName" : abcd"
1 )
At firts i created aggregation using MergeRecord with my correlation
attribute set on ${ServiceName)
the output is something like this :
[ {
"eventType" : "api",
"message" : "blabla"
}, {
"eventType" : "api",
"message" : "blabla"
}, {
"eventType" : "api",
"message" : "blabla"
}]
I then add the value to the cache with PutDistributedMapCache and cache
entry identifier on ${ServiceName}
but i'am blocked with the featching , I wish add the new aggregation
value with the values provided by the cache but if link the mergedRecord
with FetchDistributedMapCache operator will replace my value by the cache
value and i will lost the context of my entry
i'am taking a look on "Put Cache Value In Attribute"in order to keep
both context but I'm not sure if I can then merge the attributes in the
Content
Thank for your suggestion !
Yanna elina
How keep the from losing original content when a replacetext is performed for an invokehttp
I have a situation where I need to do a post using an invokehttp for performing a lock operation on another web server. This requires me to do a replacetext in order to add the required json needed for the post. I wanted to do this call before I performed a putfile call on my system but that blows away the file content I needed to put. I really want to do these to calls in this order so if the invokehttp to get the lock fails I won’t do my put. I did try using the a mergecontent before the put in order the merge the original file content with the invokehttp calls content and that seem to work with some other gymnastics. I am looking for a better if one exists.
Variables to start a NiFi flow
Hi, We have a data flow which extracts data from source database table and loads into target hive table. This flow needs to run several times in a day to get delta records from source table and also for multiple tables . Now we need to replicate this same process for all the different source tables. So rather than creating multiple data flows for each separate table, can I use the existing flow and pass parameters like source table name to that flow and the flow starts. Basically looking for an interface where the user can pass the table names that we want to load at a given point in time and the flow is triggered for that table. Variable Registry comes to mind, but I am not sure how to make it work for this use case. We are using NiFi 1.9.0 as part of CDF bundle. Thanks Bimal Mehta
Re: authenticated nifi agent wih unauthenticated registry
There is a video on the NiFi Registry page of the site that shows how to setup NiFi talking to secure NiFi Registry... https://nifi.apache.org/registry.html https://www.youtube.com/watch?v=DSO12fhnZ90=youtu.be On Mon, Aug 26, 2019 at 4:26 AM Nicolas Delsaux wrote: > > Hi all > > I have correctly setup my nifi runner to use LDAP auth from my company. > > I'm now trying to understand why registry no more work. > > As you may guess from message title, my registry is currently not > authenticated. > > Do I need to have auth enabled on registry when it is enabled on nifi > runner ? > > Thanks >
Re: AttributeRollingWindow 1.9.2 Value to Track UUID possible ?
Hi Brian ,
Thank a lot for your advice and your help :)
Steve
Le ven. 23 août 2019 à 23:36, Matt Burgess a écrit :
> Steve,
>
> In a flow-based system like NiFi it can be difficult to tell when
> there are "no more events with the same key". However if you have a
> notion of the max amount of time where events would appear, you might
> be able to use MergeRecord or MergeContent with a Correlation
> Attribute Name of "eventUuid" with a Min and Max Bin Age of that time
> interval (and make sure the Max Records is set higher than you'd ever
> expect for the number of events in the system over that time). When
> MergeRecord emits a flow file, you are assured that all the records
> have the same eventUuid, and then you might use QueryRecord to execute
> SQL on the records which allows you to do some aggregations of the
> measures.
>
> This won't be a perfect system as MergeRecord/Content only has the
> number of bins (unique eventUuids, e.g.) you specify, so it's possible
> that a single flow file does not contain all events for that UUID over
> that time period. You may be able to use an incremental pipeline like
> MergeRecord -> PartitionRecord -> MergeRecord -> QueryRecord. If
> there's a timestamp field in each record you could have something like
> "MIN(timestamp) as min_timestamp, MAX(timestamp) as max_timestamp,
> AVG(myEventMeasure) GROUP BY eventUuid" to do the aggregates. It might
> not be an exact window size but "max_timestamp - min_timestamp" would
> give you the width of the window for the records in each flow file.
>
> Queries over streams is an active research area, the folks working on
> Apache Calcite are doing some really cool things, and we've got
> Calcite powering QueryRecord under the hood, so I'm hoping we can
> continue to enjoy the fruits of their labor :)
>
> Regards,
> Matt
>
> On Fri, Aug 23, 2019 at 4:03 PM Steve Robert
> wrote:
> >
> > Hi Brian ,
> >
> > thank you for your reply.
> > To be precise I studied the possibility to open a window on an attribute
> value and close it after an user-defined time and if there are no more
> events with the same key.
> > In flink side lot operator send Event with SUCCESS OR FAIL . operator is
> running in paralleles.
> > I wanted to be able to do a aggregation to determine if there was an
> error or not during a life cycle iteration. each life cycle is unique and
> associated with a UUID
> >
> > Le ven. 23 août 2019 à 21:00, Bryan Bende a écrit :
> >>
> >> I think AttributeRollingWindow was made to aggregate a numeric
> >> attribute. So for example, if all flows files had an attribute called
> >> "amount" that was an integer, then you could say Value to Track =
> >> ${amount} and it would aggregate those values over window.
> >>
> >> In your case the attribute you have is not the value to aggregate, its
> >> a key. The value in your case is a constant of 1 where you want to
> >> increment the aggregated value by 1 whenever you see the key.
> >>
> >> I guess it depends what you want to do with the aggregated value, but
> >> your situation seems closer to a counter which could be tracked with
> >> UpdateCounter setting the Counter Name to ${eventUuid} and the Delta
> >> to 1, but then I don't know exactly how you would use this counter
> >> later.
> >>
> >> On Fri, Aug 23, 2019 at 1:59 PM Steve Robert
> >> wrote:
> >> >
> >> > Hi Guys ,
> >> >
> >> > I apologize in advance if my question seems trivial but I am new on
> Nifi.
> >> > I'm studying Nifi for an integration with Flink that I'm used to.
> >> >
> >> > Since flink I send events using Site-to-Site to Nifi.
> >> > the Flowfile have attribute
> "eventUuid":"97f82c90-0782-4aab-8850-56ee60b0b73d"
> >> >
> >> > I would like to do an aggregation on events based on the value of
> this attribute,
> >> >
> >> > so I look at the documentation of AttributeRollingWindow.
> >> >
> >> > I understood by looking at the errors that this operator does not
> accept strings and therefore can not track a value on String
> >> > Value to Track : ${eventUuid} will throw NumberFormatExceptions.
> >> >
> >> > Value to Track : ${eventUuid:toNumber()} will not work because UUID
> 128 bit can't be convert to an 32 bytes number;
> >> >
> >> > Is this a limitation or a bad approach on my part? I can not use an
> event on the timestamp here because this UUID is generated at the beginning
> of the workflow
> >> > Thank a lot for your help
> >> > Steve
> >> >
> >> >
>
Re: Reading only latest file
Tom, What about ListFile->FetchFile flowchain, ListFile configured with a maximum file age lower than the frequency of the file creation you have? Regards, Arpad On Mon, Aug 26, 2019 at 10:19 AM Tomislav Novosel wrote: > Any ideas? > Thanks, > Tom > > On Thu, 22 Aug 2019 at 16:06, Tomislav Novosel > wrote: > >> Hi all, >> >> I have scenario where I need to read only the latest(the youngest) file >> according to creation date. The files are: >> >> Load_Dump_1.001.2019-07-22_17-22-45994.ifl - creation date >> 2019-07-22T17:24:44+0200 >> Load_Dump_1.001.2019-07-22_17-25-09132.ifl - creation date >> 2019-07-22T17:26:14+0200 >> >> So I need to fetch only the second file which is the youngest. >> I have multiple folders with files and they are filtered by extension >> (.ifl). >> >> How can I filter them and fetch only the youngest .ifl file from every >> folder? >> >> BR, >> Tom >> >>
authenticated nifi agent wih unauthenticated registry
Hi all I have correctly setup my nifi runner to use LDAP auth from my company. I'm now trying to understand why registry no more work. As you may guess from message title, my registry is currently not authenticated. Do I need to have auth enabled on registry when it is enabled on nifi runner ? Thanks
Re: Reading only latest file
Any ideas? Thanks, Tom On Thu, 22 Aug 2019 at 16:06, Tomislav Novosel wrote: > Hi all, > > I have scenario where I need to read only the latest(the youngest) file > according to creation date. The files are: > > Load_Dump_1.001.2019-07-22_17-22-45994.ifl - creation date > 2019-07-22T17:24:44+0200 > Load_Dump_1.001.2019-07-22_17-25-09132.ifl - creation date > 2019-07-22T17:26:14+0200 > > So I need to fetch only the second file which is the youngest. > I have multiple folders with files and they are filtered by extension > (.ifl). > > How can I filter them and fetch only the youngest .ifl file from every > folder? > > BR, > Tom > >
