1. Introduction and Motivation for Backfilling Flink Pipelines at Netflix [1]

thumbnail
  1. The presentation will cover Flink use cases at Netflix and the need for backfilling Flink applications. [4]
  2. Existing backfilling approaches, including their pros and cons, will be discussed. [4]
  3. The technical design of the Iceberg source connector and its solution for event ordering challenges will be detailed. [5]
  4. Finally, steps to enable Iceberg backfill for Flink applications will be shared. [6]

1.1. Flink Use Cases and the Need for Backfilling [8]

  1. The speaker's team builds data systems for data analytics and machine learning algorithms. [9]
  2. One key system is the Real-Time Merge Impression app (RMI). [10]
  3. A main use case for RMI is calculating the click-through rate (CTR). [11]
  4. RMI joins impression events and playback events in real time based on a key. [13]
  5. RMI data is critical for personalization algorithms and A/B experimentation analysis. [14]
  6. RMI is one of the largest stateful Flink applications at Netflix due to large data volume. [15]
  7. Flink applications can fail for various reasons: [16]
    1. Errors when reading from source or writing to sink. [16]
    2. Dependency service failures leading to incorrect output. [17]
    3. Upstream data changes breaking Flink logic assumptions. [18]
  8. Backfilling is necessary after failures to mitigate downstream impacts on Netflix products. [19]
  9. Possible types of backfilling include: [21]
    1. Correcting wrong data previously produced by the Flink application.
    2. Recovering data for the period when the Flink job was down.
    3. Bootstrapping state to continuously consume live data from Kafka topics.
  10. Backfill involves re-running the Flink job to reprocess source events from the problematic period. [23]

1.2. Challenges with Existing Backfilling Approaches [24]

  1. Challenge 1: Kafka Retention Limits [25]
    1. Backfilling requires replaying events before source events expire due to limited Kafka retention. [25]
    2. Increasing Kafka retention is very expensive for huge datasets. [27]
    3. Retaining 30 days of data at Netflix would cost about $93 million a year. [28]
  2. Solution: Storing Kafka Events Elsewhere [29]
    1. Storing Kafka events at a lower cost for backfilling is possible. [29]
    2. Netflix Keystone platform routes Kafka events to other storage systems. [30]
    3. Events are stored in Iceberg tables on top of S3 for later batch processing. [31]

2. Iceberg as a Cost-Efficient Storage Solution for Backfilling [32]

thumbnail
  1. What is Iceberg? [32]
    1. Iceberg is an open-source table format for huge analytic datasets. [32]
    2. It offers features like schema evolution and time traveling. [33]
    3. It also provides file pruning based on partition and column level statistics. [33]
    4. Iceberg tables offer better compression rates and 98% less storage cost compared to Parquet files on S3. [34]
  2. Kafka Events in Iceberg Tables [36]
    1. Kafka events (e.g., playback events) have top-level attributes like account ID and view duration. [37]
    2. These attributes become top-level columns in the Iceberg table. [38]
    3. Kafka metadata, such as ingestion time, is automatically attached to the table. [39]
    4. This makes event data available in Iceberg at a lower cost for backfilling. [39]
  3. Challenge 2: Batch-Based Backfilling Approach [40]
    1. An existing approach is building a separate batch application (e.g., Spark job) equivalent to the Flink app. [40]
    2. This Spark job reads from the Iceberg table instead of Kafka. [40]
    3. Challenge: Initial development of the Spark job can take days or weeks. [42]
    4. Extensive data validation is needed to ensure both parallel applications produce the same results. [42]
    5. Continuous engineering effort is required to keep the Spark app up to date. [44]
    6. This batch-based approach is not ideal. [45]

3. The Iceberg Source Connector: Combining Real-Time and Batch Benefits [46]

thumbnail
  1. Summary of Common Backfilling Approaches [46]
    1. Batch-based backfill: Maintains a separate equivalent app. [47]
    2. Pro: Low-cost data retention.
    3. Con: Engineers must maintain two applications in parallel.
    4. Real-time backfilling: Re-runs the Flink application before Kafka sources expire. [48]
    5. Pro: Uses the exact same app for production and backfilling.
    6. Con: Increasing Kafka retention is very expensive.
  2. Introducing the Iceberg Source Connector [50]
    1. The Iceberg source connector combines the best aspects of both worlds. [50]
    2. Properties of the Iceberg Source Connector:
    3. Provides a generic solution for backfilling Flink applications. [52]
    4. It is easy to adopt across various Flink applications. [53]
    5. Requires very few code changes to add to an application. [54]
    6. Based on Flink's latest FLIP-27 source APIs. [55]
    7. Can scale horizontally to the Flink application's parallelism. [55]
    8. Prevents data skews on any particular reader. [55]
    9. Allows quick backfilling by adding resources to jump onto live traffic quickly. [56]
    10. It has been evaluated in production deployments, confirming its viability. [57]
  3. Mechanics of Backfilling with Iceberg Source [58]
    1. When processing from the recent past, the application reads data from the Kafka cluster (low latency). [59]
    2. When historical data (outside Kafka retention) is required, it is read from the Iceberg table. [60]
    3. Iceberg offers high throughput, allowing the backfill to complete quickly. [60]

4. Internal Design and Challenges of the Iceberg Source Connector [61]

thumbnail

4.1. Initial Assessment and FLIP-27 Implementation [61]

  1. Existing OSS Iceberg Integration [63]
    1. Existing OSS Iceberg community code supports reading Iceberg tables for bounded (backfilling) and continuous streaming. [63]
    2. However, the existing OSS source lacked critical properties: [64]
    3. It did not support stateful Flink applications expecting event ordering similar to Kafka. [65]
    4. Failure to maintain ordering would result in incorrect backfill results. [66]
    5. It was written using Flink's old source interfaces, limiting extensibility (e.g., supporting data ordering). [68]
    6. Due to these limitations, the team built their own Iceberg source connector primarily for backfilling. [69]
    7. The connector was built on FLIP-27 for better coordination between the source connector and the reader. [70]
    8. Coordination is paramount for supporting ordering and alignment between sources. [70]
  2. FLIP-27 Source API Components [72]
    1. The new source API requires implementing two separate components, unlike the old simple source function. [72]
    2. Split Enumerator: Runs on the Job Manager. [72]
    3. Responsible for discovering new splits. [72]
    4. Assigns splits to readers upon request. [72]
    5. Split: A flexible definition (e.g., Kafka partition, portion of a file). [73]
    6. Split Reader: Runs inside a Task Manager. [76]
    7. Takes a split and reads/emits records from it. [76]
  3. Iceberg Source Implementation using FLIP-27 [77]
    1. The Split Enumerator populates splits from the Iceberg table containing Kafka data. [77]
    2. The enumerator filters splits to match the backfill period. [77]
    3. When a reader requests a split, the enumerator assigns unassigned splits. [78]
    4. The reader emits records by reading the assigned split (a portion of a Parquet/ORC data file). [79]
    5. The Split Enumerator implements the SplitEnumerator interface. [82]
    6. The start method handles state initialization (e.g., pending splits). [85]
    7. The Iceberg client library filters data for the backfill period and organizes it into splits. [87]
    8. The handleSplitRequest method returns a pending split or signals no more splits are available. [85]

4.2. Challenge 1: Maintaining Event Ordering and Throughput [90]

  1. The Ordering Problem [90]
    1. Applications assume data ordering, and maintaining this is crucial for backfill correctness. [90]
    2. Example: A sessionizer operator implemented as a key process function. [94]
    3. The sessionizer tracks the start and end of a playback session. [98]
    4. It updates state and sets a timer upon receiving a new playback event. [99]
    5. The onTimer method marks the session complete if no updates occur within a threshold. [102]
    6. This logic assumes a specific order of data ingestion. [103]
    7. If splits are assigned randomly, watermarks progress incorrectly, leading to invalid output. [104]
  2. Attempt 1: Strict Injection Timestamp Ordering [107]
    1. Idea: Order splits based on their injection timestamp and assign them sequentially. [107]
    2. Example: Splits 0-5, 5-10, 10-15 are ordered by time. [109]
    3. The enumerator assigns splits 0-5 to readers first. [112]
    4. Only after completion are the next splits (5-10) assigned. [113]
  3. Problem with Strict Ordering [114]
    1. Throughput is limited by the number of splits assigned in parallel. [114]
    2. This limits the ability to throw more resources at the backfill for quick completion. [114]
    3. The solution is to relax the order constraint to address the throughput problem. [115]
  4. Attempt 2: Relaxed Ordering using Injection Watermark [116]
    1. Flink applications often use event time semantics and tolerate some lateness. [117]
    2. New State: Injection Watermark tracks the timestamp up to which data is fully processed (initially 0). [121]
    3. Lateness Constraint: Assume the application tolerates up to 10 minutes of lateness. [120]
    4. Splits in the window [Injection Watermark, Injection Watermark + Lateness] can be read in parallel. [122]
    5. Initial Assignment: Splits 0-5 and 5-10 can be read in parallel (0 + 10 = 10 minutes). [122]
    6. Splits 10-15 cannot be assigned yet, as reading them would violate the lateness constraint. [123]
    7. Only the first three available splits are assigned due to limited parallel readers. [124]
    8. Watermark Advancement: Once the first three splits are completed, the watermark moves to 5. [125]
    9. Next Assignment: Data up to 15 minutes (5 + 10) can now be read. [126]
    10. Splits 5-10 and 10-15 are assigned in parallel. [126]
    11. This approach solves the throughput problem while maintaining correctness for applications tolerating lateness. [127]

4.3. Kafka Ordering vs. Injection Watermark Ordering [128]

  1. Kafka Ordering Guarantees [129]
    1. Kafka guarantees strict ordering per Partition. [129]
    2. There are no ordering guarantees across partitions. [130]
  2. Flink Application Requirements [131]
    1. Most Netflix Flink applications use event time semantics and are written with lateness in mind. [131]
    2. For these applications, rough ordering based on injection timestamp is sufficient for correctness and performance. [132]
  3. Strict Kafka Ordering Implications [134]
    1. If strict Kafka ordering is required, Iceberg data must be partitioned along Kafka's partitioning scheme on the write path. [134]
    2. This leads to many small files, making storage exorbitantly expensive. [135]
    3. Read performance is also hurt due to the need for many IOs. [136]
    4. Use cases requiring strict Kafka ordering are outside the purview of this Iceberg source connector. [137]
    5. Netflix currently does not have such use cases. [138]

4.4. Challenge 2: Alignment for Multi-Source Applications [139]

  1. The Alignment Problem [139]
    1. Many Flink applications consume from multiple Kafka sources, requiring reading from multiple Iceberg tables during backfill. [139]
    2. If one source is significantly larger, backfilling both at the same rate causes a huge watermark skew. [140]
    3. This results in state size explosion, leading to slow checkpoints or checkpoint failures. [140]
    4. Operational difficulty might push developers back to batch applications. [141]
    5. Example: Source 2 has 2x more data/splits than Source 1. [144]
    6. Source 1 progresses at approximately 2x the rate of Source 2, causing watermark skews. [146]
  2. Solution: Global Injection Watermark Coordination [148]
    1. Idea: Coordinate between enumerators so their injection watermarks advance similarly. [148]
    2. The injection watermark state must be global, not local to each source enumerator. [149]
    3. All source enumerators coordinate with this global tracker before assigning new splits. [149]
    4. Assume 10 minutes of latency tolerance for both sources. [151]
    5. Initially, splits up to the 10th minute are assigned for both sources. [151]
    6. After Source 1 completes its splits, its enumerator cannot assign more because Source 2 is lagging. [152]
    7. The global injection watermark remains stuck at zero. [153]
    8. Source 1 waits until Source 2 completes its second set of splits (up to minute 10). [154]
    9. The global watermark is then updated to 5. [155]
    10. Both enumerators can now allocate splits up to the 15th minute, solving the alignment problem. [155]

4.5. Summary of Iceberg Source Components [156]

  1. Split Enumerator: [156]
    1. Discovers splits and assigns them to readers upon request.
    2. Notifies the watermark tracker about new splits discovered. [157]
  2. Injection Watermark Tracker: [157]
    1. Keeps track of the lowest injection watermark.
    2. Notifies the enumerator of watermark updates, allowing new split assignments.
  3. Reader: [159]
    1. Takes an assigned split.
    2. Emits records from the split.

5. Enabling and Benefits of Iceberg Backfill for Flink Applications [160]

thumbnail
  1. Enabling Iceberg Backfill (RMI Example) [161]
    1. RMI production job reads impression and playback events from Kafka topics. [162]
    2. Real-time events are also routed to Iceberg tables. [163]
    3. A backfill stack reads from the Iceberg tables and writes to the same RMI output. [163]
    4. Code Changes: Add a few lines of code to include the Iceberg source. [164]
    5. The in-memory representation of the Iceberg source is consistent with the Kafka source. [167]
    6. No changes are needed to the processing steps. [167]
    7. Configuration Changes: Define the backfill source to specify the Iceberg table. [168]
    8. A dynamic source selector is added to easily switch between live and backfill sources. [170]
  2. Operational Results and Tuning Lessons [171]
    1. The Flink UI allows easy switching to Kafka as a source. [172]
    2. For backfilling, the start and end timestamps of the backfilling window are configured. [172]
    3. The option to read from one or more regions is available. [173]
    4. Performance: RMI was able to process 24 hours of data within 5 hours. [174]
    5. Accuracy: Data validation showed the backfill output matched the production output up to 99.9%. [175]
    6. Minor discrepancies relate to RMI's deduplication logic. [176]
    7. Tuning Lessons: [178]
    8. The backfilling window depends on Flink logic. [179]
    9. RMI holds data in state for six hours and locks up to one hour of lateness. [180]
    10. To backfill RMI data, eight hours of data must be processed. [180]
    11. If event ordering is required, set the max misalignment threshold. [181]
    12. RMI set this configuration to 15 minutes. [182]
    13. Backfilling jobs need separate tuning from production jobs. [182]
    14. Backfilling jobs have a higher volume of source events per second. [182]
    15. Cluster settings should be adjusted for higher throughput and faster backfill. [183]
  3. Benefits and Roadmap [184]
    1. Benefits of Iceberg Source: [184]
    2. Uses the same Flink application for production and backfilling. [185]
    3. No changes needed to business logic. [185]
    4. Easy setup with minimal code changes. [186]
    5. Fast backfilling of large historical data due to horizontal scaling. [187]
    6. Cost-efficient: Retaining 30 days of data costs about $2 million a year in Iceberg, down from $93 million in Kafka. [188]
    7. Roadmap Items: [190]
    8. Support for continuous streaming Iceberg source for incremental updates (for applications not requiring sub-second latency). [190]
    9. Support for hybrid streaming (bootstrapping with historical Iceberg data and consuming live data). [191]
    10. Future support for applications requiring strict Kafka ordering (e.g., CDC apps). [192]