A general pattern for dynamically adding new aggregations could be something
like this
BroadcastStream<AggregationInstructions> broadcastStream =
aggregationInstructions
.broadcast(broadcastStateDescriptor);
DataStream<DataToAggregateEnrichedWithAggregationInstructions>
streamReadyToAggregate = dataToAggregate
.connect(broadcastStream)
.process(new JoinFunction())
.flatMap(new AddAggregationKeyAndDescriptor)
.keyBy('aggregationKey')
Where
* aggregationInstructions is a stream describing which fields to aggregate
by. It might contain a List<String> of the field names and another field which
can be used to describe what the aggregation is doing. Example
groupbyFields=[‘sourceIp’,’sourcePort’,’destIp’,’destPort’], groupingType =
‘bySession’, action = ‘Add’ or ‘Delete’
* JoinFunction is a KeyedBroadcastProcessFunction which adds the
groupByFields and groupingType to each message in the dataToAggregate stream
and possibly deletes groupings from state.
* AddAggregationKeyAndDescriptor is a FlatMapFunction which adds
aggregationKey to the stream based on the value of groupByFields
The flatMap means one message may be emitted several times with different
values of aggregationKey so it may belong to multiple aggregations.
From: M Singh <[email protected]>
Sent: Monday, January 24, 2022 9:52 PM
To: Caizhi Weng <[email protected]>; User-Flink <[email protected]>
Subject: Re: Apache Fink - Adding/Removing KeyedStreams at run time without
stopping the application
NOTICE: This email is from an external sender - do not click on links or
attachments unless you recognize the sender and know the content is safe.
Hi Caizhi:
Thanks for your reply.
I need to aggregate streams based on dynamic groupings. All the groupings
(keyBy) are not known upfront and can be added or removed after the streaming
application is started and I don't want to restart the application/change the
code. So, I wanted to find out, what are the options to achieve this
functionality. Please let me know if you have any advice or recommendations.
Thanks
On Monday, January 24, 2022, 04:28:32 AM EST, Caizhi Weng
<[email protected]<mailto:[email protected]>> wrote:
Hi!
Adding/removing keyed streams will change the topology graph of the job.
Currently it is not possible to do so without restarting the job and as far as
I know there is no existing framework/pattern to achieve this.
By the way, why do you need this functionality? Could you elaborate more on
your use case?
M Singh <[email protected]<mailto:[email protected]>> 于2022年1月22日周六
21:32写道：
Hi Folks:
I am working on an exploratory project in which I would like to add/remove
KeyedStreams
(https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/operators/overview/#keyby)
without restarting the Flink streaming application.
Is it possible natively in Apache Flink ? If not, is there any
framework/pattern which can be used to implement this without restarting the
application/changing the code ?
Thanks
