Hello everyone! TLDR: we are seeking guidance on how to accurately calculate the total number of records processed by a Flink job. Our Flink jobs incorporate various source operators like Kafka, JDBC, and filesystems. Once a Flink job reaches a terminal state, we aim to determine the total number of processed records, considering their statuses (e.g., success or failure). This information is crucial for calculating the job's Service Level Agreement (SLA). We are aware that the Flink REST API provides metrics such as read/write-records per operator (

/jobs/:jobid

endpoint). However, we have discovered that if a job fails (due to a Task Manager failure, for instance) and restarts from a checkpoint, the metrics counters are reset. Consequently, we cannot rely on these metrics alone. One potential approach we are considering is writing the records to a side output for each operator, including a status indicating success or failure. Subsequently, we plan to utilize a query engine like Trino to analyze the side output and accurately calculate the number of processed records. However, we are concerned about the potential overhead of writing each record to the side output for every operator. We would greatly appreciate insights from others who have tackled similar challenges. Specifically, we are interested in learning how people calculate such metrics with accuracy in mind, taking into account potential failures and without relying solely on the native metrics system. Thank you all in advance for your valuable input!

A side output can be very efficient if you do some batching, as much as you're willing to tolerate under- or over-reporting during periods when there were failures... The batches can be written into something really cheap like maybe an S3 file sink

Yeah, we are planning to write side output to s3. Also as you mentioned for over-reporting, we will have to do de-duplication. Our only concern is that data is duplicated per operator and can add quite a lot of overhead(need to do some load test to know). Btw, how can we buffer? Using window function as per https://stackoverflow.com/a/74145858 ?

Hmm - couldn’t you have a

ValueState<Long>

to accurately track the count across restarts, and expose this via a Flink metrics gauge?

I guess that would work but I'm not sure I would want state to be so load-bearing for something that smells like a billing/accounting use case