Stream Processing

3 posts

datadog3 min readCurated summary

How we measure data completeness at scale

Datadog built a real-time data-completeness system to ensure that every customer’s telemetry is available for dashboards, alerts, queries, and AI-driven decisions. Because ingestion spans hundreds of distributed paths and customers may send delayed or retried data, global or watermark-based tracking is unreliable. The system instead tracks payloads segment by segment, using idempotent create and acknowledgment events to identify losses and calculate end-to-end completeness. ## Defining Completeness at Datadog’s Scale - Completeness means every ingested payload—metrics, logs, spans, or other telemetry—is ultimately available to customers. - The system must measure completeness: - Across hundreds of services and ingestion paths - For each individual customer - In real time - With enough detail to identify where degradation occurred - Customer traffic may take different routes because of partitioning, isolation, and traffic patterns. - Metrics and APM pipelines can each involve hundreds or tens of distinct paths, creating a large number of possible failure points. - The completeness system must remain independent of the services it monitors so it can provide trustworthy diagnostics during incidents. ## Tracking Completeness by Pipeline Segment - Datadog considered watermark-based tracking, but delayed customer data, replayed traffic, and pipeline loops made predictable watermarks impractical. - Pipelines are divided into segments representing steps within or between services. - For example, intake-in to intake-out is one segment. - Intake-out to processing-in is another. - Each segment is measured independently, allowing engineers to locate degradation within a service or between services. - Segment-level tracking also adapts to pipelines whose branches appear or disappear over time. ## Counting Creates and Acknowledgments - When a payload enters a segment, the system records a create event. - When it exits, the system records an acknowledgment using the payload’s unique identifier. - Comparing creates with acknowledgments reveals whether payloads were lost in that segment. - Events are organized into time buckets based on when the payload first entered Datadog, using a Datadog-controlled timestamp rather than the customer’s clock. - Each identifier has a state per segment: - Created - Acknowledged - Acknowledged before the create event arrived - Duplicate create or acknowledgment events are ignored, making the system idempotent despite retries and event reordering. ## Calculating End-to-End Completeness - Segment completeness is the ratio of payloads exiting a segment to those entering it. - For sequential services, overall completeness is calculated by multiplying segment ratios. - Parallel branches require a different approach: - Treating branches as one pipeline would make completeness wait for the slowest branch. - Instead, Datadog uses a weighted average, giving each branch influence proportional to the volume it processes. - In the example, one branch reaches 94% completeness by multiplying 98% and 96% across two sequential services, while another branch reaches 100%. - Combining these branch measurements produces a more accurate view of currently available data without incorrectly marking all data incomplete because one branch is slower. ## Practical Conclusion Segment-level, identifier-based tracking gives Datadog a real-time and customer-specific view of data completeness. It both supports reliable end-to-end calculations and helps humans or automated systems quickly determine where ingestion problems are occurring.

Read original(opens in new tab)
toss5 min readCurated summary

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)
netflixOriginal article

How and Why Netflix Built a Real-Time Distributed Graph: Part 1 — Ingesting and Processing Data Streams at Internet Scale | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix has developed a Real-Time Distributed Graph (RDG) to unify member interaction data across its expanding business verticals, including streaming, live events, and mobile gaming. By transitioning from siloed microservice data to a graph-based model, the company can perform low-latency, relationship-centric queries that were previously hindered by expensive manual joins and data fragmentation. The resulting system enables Netflix to track user journeys across various devices and platforms in real-time, providing a foundation for deeper personalization and pattern detection. ### Challenges of Data Isolation in Microservices * While Netflix’s microservices architecture facilitates independent scaling and service decomposition, it inherently leads to data isolation where each service manages its own storage. * Data scientists and engineers previously had to "stitch" together disparate data from various databases and the central data warehouse, which was a slow and manual process. * The RDG moves away from table-based models to a relationship-centric model, allowing for efficient "hops" across nodes without the need for complex denormalization. * This flexibility allows the system to adapt to new business entities (like live sports or games) without requiring massive schema re-architectures. ### Real-Time Ingestion and Normalization * The ingestion layer is designed to capture events from diverse upstream sources, including Change Data Capture (CDC) from databases and request/response logs. * Netflix utilizes its internal data pipeline, Keystone, to funnel these high-volume event streams into the processing framework. * The system must handle "Internet scale" data, ensuring that events from millions of members are captured as they happen to maintain an up-to-date view of the graph. ### Stream Processing with Apache Flink * Netflix uses Apache Flink as the core stream processing engine to handle the transformation of raw events into graph entities. * Incoming data undergoes normalization to ensure a standardized format, regardless of which microservice or business vertical the data originated from. * The pipeline performs data enrichment, joining incoming streams with auxiliary metadata to provide a comprehensive context for each interaction. * The final step of the processing layer involves mapping these enriched events into a graph structure of nodes (entities) and edges (relationships), which are then emitted to the system's storage layer. ### Practical Conclusion Organizations operating with a highly decoupled microservices architecture should consider a graph-based ingestion strategy to overcome the limitations of data silos. By leveraging stream processing tools like Apache Flink to build a real-time graph, engineering teams can provide stakeholders with the ability to discover hidden relationships and cross-domain insights that are often lost in traditional data warehouses.