Extending Real-time Ad Frequency Capping Aggregation to One Week with Apache Flink + RocksDB Tuning
The post describes Toss’s expansion of real-time advertising frequency-capping from short Flink windows to periods of up to seven days. The new system provides accurate sliding counts from one minute to seven days through a single Redis lookup, while treating Flink state as the authoritative source and Redis as its projection. The migration addressed architectural complexity, backfill consistency, and distinct RocksDB bottlenecks across three specialized Flink applications. ## Frequency Capping and Its Business Impact - Frequency capping controls how many times an individual user sees an advertisement. - Incorrect counts can: - Waste an advertiser’s budget through excessive exposure. - Prevent valid impressions when the system believes a limit has already been reached. - Different products require different windows, such as: - Three impressions per day. - One impression over the previous seven days. - The target system therefore needed accurate, real-time sliding counts from one minute through seven days. ## Limitations of the Previous Batch-Oriented System The original architecture combined three Airflow-managed layers: - **Head** - Stored current-day and previous-day events in Redis through a Spring Kafka consumer. - Updated counts immediately per event. - **Mid** - Used daily Spark jobs to pre-aggregate data from D-2 through D-7. - **Tail** - Added hourly correction data around the boundary between Head and Mid. - Airflow workflows ran approximately 75 times per day. At serving time, the API could perform up to four Redis lookups and combine the results. - This structure was difficult to maintain because of the dependencies and boundary conditions between Head, Mid, and Tail. - Time-based truncation made precise event-level sliding windows difficult. - The architecture remains useful for longer windows such as 30 days and fixed daily aggregates, especially when data exceeds Kafka retention and must be recovered from batch storage. - Extending the existing short-window Flink system was chosen to simplify serving and reduce DAG complexity. ## Three Flink Applications Rather than place all windows in one Flink job, the team split processing into three applications with shared code but independent RocksDB configurations: - **Minutes** - Handles one- to 30-minute windows. - Frequent event expiration creates heavy write traffic. - Its main concern is RocksDB Write Buffer Manager pressure and resulting Write Stalls. - **Hours** - Handles windows up to 12 hours. - Maintains many more advertisement IDs in state. - Filter Block Cache misses can saturate CPU. - Redis synchronization requires an O(N) scan over advertisement IDs in each window. - Filter Block tuning and additional managed memory are important. - **Days** - Handles the largest state volume. - A seven-day window can produce approximately 68 GB of live SST files and 220–230 GB savepoints. - Checkpoint I/O becomes the primary bottleneck, motivating a Flink Changelog design. Separating the applications allowed each workload’s RocksDB and runtime bottlenecks to be optimized independently without affecting the others. ## Backfill and Catch-up Architecture The most difficult migration problem was maintaining correctness at the transition point between historical data and live processing. - **Backfill** - Loads seven days of historical events. - Only increments counts. - Does not register expiration timers. - Synchronizes the initialized values to Redis once and then finishes. - **Catch-up** - Re-reads historical events from Kafka. - Rebuilds both counts and expiration timers. - Begins writing to Redis after reaching the historical scan end. - Enables each window only after sufficient lookback data has been reconstructed. The two phases cannot safely share one pipeline: - Backfill must only add historical counts. - Live or catch-up processing must both add new events and subtract events that leave the sliding window. - If expiration timers ran while backfill was incomplete, decrements could occur before all historical increments had been applied, producing incorrect results. - Flink batch mode was rejected because state is discarded when the job finishes. - A Spark and Hive-based approach was also rejected because it would introduce additional systems and complicate the single-source-of-truth model. Separate Kafka consumer groups were required so that backfill offsets would not cause catch-up events to be skipped. ## State as the Single Source of Truth - Flink state stores the authoritative aggregate. - Redis is treated only as a serving projection. - If Redis becomes inconsistent, it can be reconstructed from Flink state. - This design preserves correctness during failures, restarts, and Redis resynchronization. ## Maintaining Transition Consistency Three mechanisms were combined to make the backfill-to-catch-up boundary reliable: - **Redis write condition** - Writes are based on each event’s `eventTime` being after the backfill completion point. - Using the global watermark directly could block all writes because one slow or idle partition can hold back the watermark. - **`withIdleness` set to 60 seconds** - Excludes inactive Kafka partitions from watermark progression. - A longer timeout avoids falsely marking a partition idle just before a bounded source emits `MAX_WATERMARK`. - **Timer state TTL** - Must exceed the sliding-window expiration period. - If the timer fires after its associated state has expired, `timerState.get()` returns null and the decrement is skipped. - This would leave counts artificially high after delays or recovery. - The state is manually cleaned up after timer processing. ## RocksDB and Flink Runtime Tuning Once the system was serving real-time results, operational metrics exposed different bottlenecks in each application. - The minutes application initially experienced RocksDB Write Stalls caused by pressure on the shared Write Buffer Manager. - RocksDB first stores writes in MemTables and flushes them into SST files organized across levels L0–L6. - Flink maps managed state types such as `MapState` and `ValueState` to separate RocksDB Column Families. - Because multiple Column Families share the Write Buffer Manager’s memory budget, write-heavy workloads require careful tuning of RocksDB memory and write paths. - The hours and days applications require different optimizations focused on cache misses, CPU usage, checkpoint I/O, and level management. ## Practical Conclusion For real-time frequency capping, a unified Flink-based design can simplify serving and improve sliding-window accuracy, but long windows should not automatically be combined with short ones in a single job. Separate applications, state-as-SSOT, distinct backfill and catch-up pipelines, and workload-specific RocksDB tuning are essential for maintaining correctness and operability at scale.
Read original(opens in new tab)