How to aggregate on different keys in single Flink pipeline

user3553913 0
Viewed 20

We have requirement where we need aggregate the data over two different keys. Input record for the job is as below.

{
   .....
   .....
   "key1" : "093570lells99345dfklsfkd",
   "key2" : "8587656783487535nxdghljd",
   .....
   .....
}

The records are stored in couchbase. Now Flink data pipeline would look like.

env.addSource(readFromCouchBase...)
  .name("couchbase-flinkjob")
  .assignTimestampsAndWatermarks(
    new TimestampExtractorAndWatermarkEmitter(60 * 1000, false))
  .keyBy[(String)](key1)
  .timeWindow(Time.seconds(10))
  .aggregate(new CountGroupFunctionWithEventTimeProcessing, new CountGroupWindowFunction)
  .addSink(new IdempotentPostgresSqlSinkFunction).name("postgres-sink")

Now we need grouping to happen on key2 as well. For this do we need to create new pipeline or is there a way such that we can modify the current pipeline to facilitate groupings to happen separately on key1 and key2? Efficient way would be to use the current pipeline itself to get the required groupings - as this would avoid reading from the couchbase twice.

data-science apache-flink
1 Answers

You can do that like this:

val events = env.addSource(readFromCouchBase...)
  .name("couchbase-flinkjob")
  .assignTimestampsAndWatermarks(
    new TimestampExtractorAndWatermarkEmitter(60 * 1000, false));

events
  .keyBy[String](key1)
  .timeWindow(Time.seconds(10))
  .aggregate(new CountGroupFunctionWithEventTimeProcessing, new CountGroupWindowFunction)
  .name("window-for-key1")
  .addSink(new IdempotentPostgresSqlSinkFunction)
  .name("postgres-sink");

events
  .keyBy[String](key2)
  ...
David Anderson0
Related