Add tiled implementation of the Flink app #627

caiocamatta-stripe · 2023-11-30T23:20:46Z

Summary

Adds a tiled implementation of the Flink app.

The tiled version of the Flink job adds a window / tiling operator. It has roughly the following DAG:

Kafka source - Reads objects of type T (specific case class, Thrift / Proto) from a Kafka topic
Spark expression eval - Evaluates the Spark SQL expression in the GroupBy and projects and filters the input data
Window/tiling - This window aggregates incoming events, keeps track of the IRs, and sends them forward so they are written out to the KV store
Avro conversion - Finishes converting the output of the window (the IRs) to a form that can be written out to the KV store (PutRequest object)
KV store writer - Writes the PutRequest objects to the KV store using the AsyncDataStream API

In a subsequent PR I will add documentation on tiling in general -- probably in a markdown file. This documentation will be similar to the "Flink and Tiling" presentation I gave at the Stripe <> AirBnB summit we had.

Why / Goal

In our experience at Stripe, the tiled version of Chronon (Flink + Fetcher) has much lower latency and significantly better scalability than the untiled/regular version.

Test Plan

Added Unit Tests
Integration tested

I added

an end-to-end test for the untiled app that confirms the events are processed correctly.
tests for the key selector function
tests for the row aggregators

Checklist

[~] Documentation update
- Comments in the code serve as immediate documentation, but as mentioned before I will be adding additional documentation

Reviewers

@piyushn-stripe @nikhilsimha @cristianfr

piyushn-stripe

Largely looks good, few minor comments

piyushn-stripe · 2024-01-02T14:04:36Z

flink/src/main/scala/ai/chronon/flink/AvroCodecFn.scala

-    extends RichFlatMapFunction[Map[String, Any], PutRequest] {
-  @transient lazy val logger = LoggerFactory.getLogger(getClass)
+// This utility contains common code for AvroCodecFn and TiledAvroCodecFn
+sealed trait AvroCodecFnUtility {


how about a name like BaseAvroCodecFn ? I tend to associate utility with singletons which isn't what you're getting at here

to make things explicit, we could make this an abstract class that extends from RichFlatMapFunction (so then subclasses know their contract is to provide groupByServingInfoParsed and fill out the open / flatMap methods)

agreed - BaseAvroCodecFn is a better name and making the contracts clearer is a good idea.

piyushn-stripe · 2024-01-02T14:05:12Z

flink/src/main/scala/ai/chronon/flink/AvroCodecFn.scala

-case class AvroCodecFn[T](groupByServingInfoParsed: GroupByServingInfoParsed)
-    extends RichFlatMapFunction[Map[String, Any], PutRequest] {
-  @transient lazy val logger = LoggerFactory.getLogger(getClass)
+// This utility contains common code for AvroCodecFn and TiledAvroCodecFn


lets also highlight what the subclasses need to override / specialize

piyushn-stripe · 2024-01-02T14:13:50Z

flink/src/main/scala/ai/chronon/flink/FlinkJob.scala

-  *  Spark expression eval - Evaluates the Spark SQL expression in the GroupBy and projects and filters the input data
-  *  Avro conversion - Converts the Spark expr eval output to a form that can be written out to the KV store (PutRequest object)
-  *  KV store writer - Writes the PutRequest objects to the KV store using the AsyncDataStream API
+  *   Kafka source -> Spark expression eval -> Avro conversion -> KV store writer


this changes now with the new tiling operators right? Should we add those too or maybe we move the operator breakdown to the run* methods

piyushn-stripe · 2024-01-02T14:23:32Z

flink/src/test/scala/ai/chronon/flink/test/FlinkJobIntegrationTest.scala

@@ -125,11 +88,76 @@ class FlinkJobIntegrationTest {
    // capture the datastream of the 'created' timestamps of all the written out events
    val writeEventCreatedDS = CollectSink.values.asScala

+    println(writeEventCreatedDS.size)


ezvz · 2024-01-10T03:04:56Z

aggregator/src/main/scala/ai/chronon/aggregator/windowing/Resolution.scala

+    * Find the smallest tail window resolution in a GroupBy. Returns None if the GroupBy does not define any windows.
+    * The window resolutions are: 5 min for a GroupBy a window < 12 hrs, 1 hr for < 12 days, 1 day for > 12 days.
+    * */
+  def getSmallestWindowResolutionInMillis(groupBy: GroupBy): Option[Long] =


This could live in Extensions - we already have helper functions like MaxWindow etc on GroupBy as implicits: https://github.com/airbnb/chronon/blob/master/api/src/main/scala/ai/chronon/api/Extensions.scala#L416

Extensions does seem like a good place for this, but it would require api to depend on aggregator so it can use FiveMinuteResolution.calculateTailHop. Or duplicating calculateTailHop in Extensions. What do you think?

Aggregator should already be able to access api/Extensions.scala

@caiocamatta-stripe if IntelliJ doesn't auto-complete or detect the functions in Aggregator you might need to install a scala extension, because the implicits need a little help to work in IDE I think. But it should still compile from sbt.

Hmm, I may be doing something silly..

error message: object aggregator is not a member of package ai.chronon

code diff

Feel free to move resolution & 5minResolution into api if you want to achieve this. But not necessary in this PR IMO

ezvz

Deleting my last comment, I see that you did create some tests. Will continue reviewing shortyl!

ezvz · 2024-01-10T03:19:44Z

flink/src/main/scala/ai/chronon/flink/FlinkJob.scala

+    // We use Flink "Side Outputs" to track any late events that aren't computed.
+    val tilingLateEventsTag = OutputTag[Map[String, Any]]("tiling-late-events")
+
+    // The tiling operator works the following way:


Nice comments!

ezvz

Everything looks pretty good to me. Might be good to get a stamp from @nikhilsimha too for this one.

ezvz · 2024-01-16T19:34:08Z

flink/src/test/scala/ai/chronon/flink/test/FlinkJobIntegrationTest.scala

+      List("id1") -> List(4.0), // Add up the double_val of the two 'id1' events
+      List("id2") -> List(10.0)
+    )
+


It does not necessarily need to be part of this PR, but eventually having a test case with events that fall outside of the tail of the window might be interesting to make sure that we tail-exclude properly.

Good idea! cc @piyushn-stripe - one of us can include it when we get a chance.

Yeah +1 to that. Would be a good follow up (for both the tiled & untiled)

…flink-tiling-oss

caiocamatta-stripe · 2024-02-14T14:50:46Z

hey @nikhilsimha can I get your review too pls?

aggregator/src/main/scala/ai/chronon/aggregator/windowing/Resolution.scala

nikhilsimha · 2024-02-27T03:59:24Z

flink/src/main/scala/ai/chronon/flink/AsyncKVStoreWriter.scala

@@ -75,6 +75,8 @@ class AsyncKVStoreWriter(onlineImpl: Api, featureGroupName: String)
  // The context used for the future callbacks
  implicit lazy val executor: ExecutionContext = AsyncKVStoreWriter.ExecutionContextInstance

+  // One may want to use different KV stores depending on whether tiling is on.
+  // The untiled version of Chronon works on "append" store semantics, and the tiled version works on "overwrite".


nikhilsimha · 2024-02-27T04:01:03Z

flink/src/main/scala/ai/chronon/flink/AvroCodecFn.scala

+  * @tparam IN The input data type which contains the data to be avro-converted to bytes.
+  * @tparam OUT The output data type (generally a PutRequest).
  */


I think the naming itself makes a lot of what is written here obvious.

I am personally feel auto generated doc strings reduce readability of code, but it is subjective.

I don't think it hurts here, but noted!

I'll leave it for now but don't mind removing it if anyone has strong opinions.

flink/src/main/scala/ai/chronon/flink/AvroCodecFn.scala

nikhilsimha · 2024-02-27T04:03:24Z

flink/src/main/scala/ai/chronon/flink/FlinkJob.scala

@@ -47,7 +58,25 @@ class FlinkJob[T](eventSrc: FlinkSource[T],
  // The source of our Flink application is a Kafka topic
  val kafkaTopic: String = groupByServingInfoParsed.groupBy.streamingSource.get.topic

+  /**
+    * The "untiled" version of the Flink app.


I really like this style of comments - over the autogen doc strings

nikhilsimha · 2024-02-27T04:03:58Z

flink/src/main/scala/ai/chronon/flink/window/FlinkRowAggregators.scala

+        if (debug) {
+          logger.info(


why not logger.debug?

The idea was to copy what's done in BaseFetcher, but I do prefer logger.debug

I changed to logger.debug, I prefer it that way too

nikhilsimha

Solid PR!

Can you also add a Readme to the chronon/flink dir with a rough control flow diagram?

caiocamatta-stripe · 2024-02-27T19:35:16Z

@nikhilsimha, I can add a control flow in a follow-up PR. For now we have the doc on the Tiled Architecture (which I'm merging along with this PR).

caiocamatta-stripe added 26 commits December 13, 2023 09:43

Add custom triggers

a202daf

Move triggers

3b894a6

Add KeySelector

393d58b

Comments

28aa0ab

Rename tiling package to window

27295e7

WIP runTiledGroupByJob

1224965

Comment-out AsyncKVStoreWriterTest.scala ? question mark ?

4b73485

Add ChrononFlinkRowAggregators

edac39f

Refactor AvroCodec slightly

083b870

Add TiledAvroCodecFn

a39496d

Add LateEventCounter

d4816bc

Finish runTiledGroupByJob

4f295c8

Add ChrononFlinkRowAggregationFunctionTest

321eaaa

Add missing @test decorator

e2e85df

Add KeySelector tests

369c62e

Add e2e tiled test

75ac625

Scalafmt

d0c35b1

Comments

68eb5bc

Uncomment AsyncKVStoreWriterTest

e6ebf3a

Remove slot sharing so that test finally halts

40d9a08

Tweak strings in key selector test

213cb7b

Rename files, change comments

09c305b

keyToBytes in process function should convert to array first

d52e0fc

Refactor tiled Flink test, use watermark strategy

b249063

Improve e2e test so that we check actual tile IRs

e66b17e

rm debug=true

920cc4e

caiocamatta-stripe force-pushed the caiocamatta--add-flink-tiling-oss branch from 79eeecf to 920cc4e Compare December 13, 2023 14:47

caiocamatta-stripe added 3 commits December 13, 2023 09:54

Use log4j

f10468c

Remove comment

164fc62

Minor clean up, change comments

85431e1

caiocamatta-stripe requested a review from cristianfr December 15, 2023 00:35

piyushn-stripe reviewed Jan 2, 2024

View reviewed changes

caiocamatta-stripe added 3 commits January 3, 2024 10:18

Make BaseAvroCodecFn abstract

30f2cd7

Update FlinkJob comments

829a914

Comment

2128cfe

caiocamatta-stripe requested a review from piyushn-stripe January 3, 2024 15:57

piyushn-stripe approved these changes Jan 4, 2024

View reviewed changes

ezvz reviewed Jan 10, 2024

View reviewed changes

ezvz approved these changes Jan 16, 2024

View reviewed changes

caiocamatta-stripe added 4 commits January 16, 2024 18:28

Move getSmallestWindowResolutionInMillis to GroupByOps

e114e18

Use new GroupByOps method, fix mistake

b93e292

Merge remote-tracking branch 'upstream/master' into caiocamatta--add-…

54337a3

…flink-tiling-oss

Revert "Move getSmallestWindowResolutionInMillis to GroupByOps"

b2c6716

nikhilsimha reviewed Feb 27, 2024

View reviewed changes

aggregator/src/main/scala/ai/chronon/aggregator/windowing/Resolution.scala Outdated Show resolved Hide resolved

nikhilsimha reviewed Feb 27, 2024

View reviewed changes

aggregator/src/main/scala/ai/chronon/aggregator/windowing/Resolution.scala Outdated Show resolved Hide resolved

nikhilsimha reviewed Feb 27, 2024

View reviewed changes

flink/src/main/scala/ai/chronon/flink/AvroCodecFn.scala Outdated Show resolved Hide resolved

nikhilsimha reviewed Feb 27, 2024

View reviewed changes

nikhilsimha approved these changes Feb 27, 2024

View reviewed changes

caiocamatta-stripe added 2 commits February 27, 2024 14:24

Use toScala, use multiline strings

c72ee4e

Use logger.debug

c366012

caiocamatta-stripe added 2 commits February 27, 2024 16:12

Scalafmt

aab3496

Fix compile error

2f53c7e

caiocamatta-stripe merged commit dcb3750 into main Feb 28, 2024
7 checks passed

caiocamatta-stripe deleted the caiocamatta--add-flink-tiling-oss branch February 28, 2024 17:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add tiled implementation of the Flink app #627

Add tiled implementation of the Flink app #627

caiocamatta-stripe commented Nov 30, 2023 •

edited

Loading

piyushn-stripe left a comment

piyushn-stripe Jan 2, 2024

piyushn-stripe Jan 2, 2024

caiocamatta-stripe Jan 3, 2024

piyushn-stripe Jan 2, 2024

piyushn-stripe Jan 2, 2024

piyushn-stripe Jan 2, 2024

ezvz Jan 10, 2024

caiocamatta-stripe Jan 15, 2024

nikhilsimha Jan 16, 2024

ezvz Jan 16, 2024

caiocamatta-stripe Jan 16, 2024

nikhilsimha Feb 27, 2024

ezvz left a comment •

edited

Loading

ezvz Jan 10, 2024

ezvz left a comment

ezvz Jan 16, 2024

caiocamatta-stripe Jan 16, 2024

piyushn-stripe Jan 17, 2024

caiocamatta-stripe commented Feb 14, 2024

nikhilsimha Feb 27, 2024

nikhilsimha Feb 27, 2024

caiocamatta-stripe Feb 27, 2024

nikhilsimha Feb 27, 2024

nikhilsimha Feb 27, 2024

caiocamatta-stripe Feb 27, 2024

caiocamatta-stripe Feb 27, 2024

nikhilsimha left a comment

caiocamatta-stripe commented Feb 27, 2024

Add tiled implementation of the Flink app #627

Add tiled implementation of the Flink app #627

Conversation

caiocamatta-stripe commented Nov 30, 2023 • edited Loading

Summary

Why / Goal

Test Plan

Checklist

Reviewers

piyushn-stripe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ezvz left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ezvz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

caiocamatta-stripe commented Feb 14, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nikhilsimha left a comment

Choose a reason for hiding this comment

caiocamatta-stripe commented Feb 27, 2024

caiocamatta-stripe commented Nov 30, 2023 •

edited

Loading

ezvz left a comment •

edited

Loading