Apache Druid

4 posts

netflix4 min readCurated summary

Stop Answering the Same Question Twice: Interval-Aware Caching for Druid at Netflix Scale

Netflix’s Druid deployment now exceeds 10 trillion rows and can ingest 15 million events per second, but repetitive dashboard queries became a scaling problem. Its new experimental caching layer handles rolling time windows by reusing settled historical results and querying Druid only for recent, changing data. Netflix accepts up to five seconds of additional staleness in exchange for substantially lower query load. ## The Scaling Problem - A dashboard with 26 charts can issue 64 queries per load. - Viewed by 30 people and refreshed every 10 seconds, that becomes roughly 192 queries per second. - Druid’s full-result cache misses whenever a rolling time interval changes. - Druid avoids caching realtime segments to preserve result correctness and determinism. - Per-segment caching reduces historical scans but still requires brokers to gather and merge results for every request. - Adding hardware to handle this redundant workload would be prohibitively expensive. ## Caching Only the Unsettled Data - In a three-hour query, most data is already stable; only the newest minutes are likely to change. - The cache stores previously returned historical portions and sends Druid only the uncached interval. - This approach is designed for time-grouped queries such as timeseries and groupBy queries. ## Deliberate Staleness - The cache can make the newest data up to five seconds stale. - This is acceptable because dashboards typically refresh every 10–30 seconds. - Netflix’s pipeline already has up to roughly five seconds of latency at P90. - Many queries also intentionally end at `now-1m` or `now-5s` to avoid unstable, newly arriving data. ## Exponential TTLs - Cache lifetimes increase with the age of each data point because older data is less likely to change. - Data under two minutes old has a minimum TTL of five seconds. - After that, TTL doubles for each additional minute: - 10 seconds at two minutes old - 20 seconds at three minutes - 40 seconds at four minutes - TTLs are capped at one hour. - Fresh data is refreshed frequently to account for late-arriving events, while older data remains cached longer. ## Time-Based Bucketing - A single cache entry per query and interval would still miss whenever a rolling window shifted. - Netflix instead uses a map-of-maps: - The outer key is a hash of the query excluding its time interval. - Inner keys represent timestamps bucketed by query granularity or at least one minute. - Big-endian timestamp encoding preserves chronological order for efficient range scans. - A three-hour query at one-minute granularity becomes 180 independently cached buckets. - When the window moves, most buckets can be reused and only the newly exposed range must be fetched. ## Router-Integrated Cache Service - The cache currently operates as an external service behind the Druid Router. - Cacheable requests are intercepted transparently: - Fully cached requests are answered directly. - Partially cached requests are narrowed to the missing interval and sent to Druid. - Metadata queries and queries without time-based grouping bypass the cache. - The proxy can be enabled or disabled without changing clients. - Netflix views this as an interim design while exploring deeper integration with Druid. ## Query Identification and Lookup - Incoming queries are parsed to extract their interval, granularity, and structure. - A SHA-256 hash is generated from the query’s logical contents, including datasource, filters, aggregations, and relevant context properties, while excluding the time interval. - The cache looks for buckets within the requested range. - Lookup requires cached buckets to be contiguous from the beginning of the requested interval; the provided article text ends while explaining the handling of expired or missing buckets. Netflix’s approach is best suited to frequently repeated rolling-window dashboards where a small, slightly stale tail is acceptable. Segmenting results by time and assigning age-based TTLs allows the system to preserve freshness where it matters while eliminating most redundant Druid work.

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

Osprey: Open Sourcing our Rule Engine

Discord is open-sourcing Osprey, a rule engine designed to help platforms detect and respond to emerging safety threats in real time. Built with ROOST and internet.dev, it processes platform events, evaluates configurable rules, and produces actionable verdicts with minimal engineering effort. Osprey emphasizes scale, rapid rule deployment, transparency, extensibility, and continuous improvement. ## Goals for a Modern Rule Engine Osprey was designed around several requirements: - Process thousands of events per second in real time. - Let teams create and deploy expressive rules within minutes. - Return clear verdicts indicating whether activity is safe, suspicious, or malicious. - Explain how rules were executed and expose errors for investigation and debugging. - Support feedback loops that improve future detection rules. - Remain extensible enough to address new attack patterns. ## Osprey’s Processing Model Osprey accepts platform events called **Actions** through either: - Synchronous gRPC requests. - Asynchronous message queues. The engine evaluates these actions using rules written in SML, a Python-based rule language. Rules can use Python UDFs, Features, and Effects, while synchronous requests can return Verdict effects directly to callers. Outputs are sent to Apache Druid, which powers investigation and analysis tools. ## Actions Actions are JSON-like events submitted to Osprey. - Each action type has a unique name and schema. - Callers can customize the payload with relevant platform data. - Example data includes login attempts, user IDs, usernames, email addresses, and IP addresses. - Rules extract and evaluate values from these action payloads. ## Rules and SML Rules are the central mechanism for detecting suspicious behavior. - SML uses a Python-inspired syntax intended to be accessible to less-technical rule authors. - Rules can reference other rules and extracted data. - Static validation enforces consistent rule-writing practices. - Validation can be extended with Python, from naming conventions to more complex domain-specific checks. - Example rules identify a known spammer by email and apply a `spammer` label to the associated user entity. ## User-Defined Functions UDFs are regular Python functions that extend Osprey’s rule language and standard library. - Built-in capabilities such as `Rule`, `WhenRules`, and `JsonData` are implemented as UDFs. - Teams can add their own UDFs when integrating Osprey into other products. - UDFs can retrieve information from external services, including machine-learning models. - They can be configured for asynchronous execution and access external-service providers through the execution context. - A sample UDF obtains a link-spam score from an external prediction service. ## Features and Entities Features are globally named variables produced during Osprey executions. - Features are exported to Apache Druid for later querying and investigation. - Prefixing a variable name with `_` keeps it local instead of exporting it. - Examples include `UserId` and `UserEmail`, extracted from JSON action data. - Entities are a specialized type of Feature representing persistent objects such as users, servers, or email addresses. - Entities can receive effects such as labels, classifications, and signals. - Entity types determine which effects are valid through static validation. - The Osprey interface provides dedicated Entity Views for examining an entity’s history. ## Effects Effects are outcomes triggered when rules evaluate as true. - They are validated and processed in aggregate after execution. - Effects can modify or annotate entities with labels, classifications, or signals. - Verdict effects can be returned synchronously to inform the requesting service of a safety determination. Osprey’s open-source release gives platforms a reusable foundation for real-time trust and safety enforcement. Teams interested in adopting it can explore the repository at [github.com/roostorg/osprey](https://github.com/roostorg/osprey).

Read original(opens in new tab)
tossOriginal article

Customers Never Wait: How to Skyrocket (opens in new tab)

Toss Payments addressed the challenge of serving rapidly growing transaction data within a microservices architecture (MSA) by evolving their data platform from simple Elasticsearch indexing to a robust CQRS pattern. While Apache Druid initially provided high-performance time-series aggregation and significant cost savings, the team eventually integrated StarRocks to overcome limitations in data consistency and complex join operations. This architectural journey highlights the necessity of balancing real-time query performance with operational scalability and domain decoupling. ### Transitioning to MSA and Early Search Solutions * The shift from a monolithic structure to MSA decoupled application logic but created "data silos" where joining ledgers across domains became difficult. * The initial solution utilized Elasticsearch to index specific fields for merchant transaction lookups and basic refunds. * As transaction volumes doubled between 2022 and 2024, the need for complex OLAP-style aggregations led to the adoption of a CQRS (Command Query Responsibility Segregation) architecture. ### Adopting Apache Druid for Time-Series Data * Druid was selected for its optimization toward time-series data, offering low-latency aggregation for massive datasets. * It provided a low learning curve by supporting Druid SQL and featured automatic bitmap indexing for all columns, including nested JSON keys. * The system decoupled reads from writes, allowing the data team to serve billions of records without impacting the primary transaction databases' resources. ### Data Ingestion: Message Publishing over CDC * The team chose a message publishing approach via Kafka rather than Change Data Capture (CDC) to minimize domain dependency. * In this model, domain teams publish finalized data packets, reducing the data team's need to maintain complex internal business logic for over 20 different payment methods. * This strategy simplified system dependencies and leveraged Druid’s ability to automatically index incoming JSON fields. ### Infrastructure and Cost Optimization in AWS * The architecture separates computing and storage, using AWS S3 for deep storage to keep costs low. * Performance was optimized by using instances with high-performance local storage instead of network-attached EBS, resulting in up to 9x faster I/O. * The team utilized Spot Instances for development and testing environments, contributing to a monthly cloud cost reduction of approximately 50 million KRW. ### Operational Challenges and Druid’s Limitations * **Idempotency and Consistency:** Druid struggled with native idempotency, requiring complex "Merge on Read" logic to handle duplicate messages or state changes. * **Data Fragmentation:** Transaction cancellations often targeted old partitions, causing fragmentation; the team implemented a 60-second detection process to trigger automatic compaction. * **Join Constraints:** While Druid supports joins, its capabilities are limited, making it difficult to link complex lifecycles across payment, purchase, and settlement domains. ### Hybrid Search and Rollup Performance * To ensure high-speed lookups across 10 billion records, a hybrid architecture was built: Elasticsearch handles specific keyword searches to retrieve IDs, which are then used to fetch full details from Druid. * Druid’s "Rollup" feature was utilized to pre-aggregate data at ingestion time. * Implementing Rollup reduced average query response times from tens of seconds to under 1 second, representing a 99% performance improvement for aggregate views. ### Moving Toward StarRocks * To solve Druid's limitations regarding idempotency and multi-table joins, Toss Payments began transitioning to StarRocks. * StarRocks provides a more stable environment for managing inconsistent events and simplifies the data flow by aligning with existing analytical infrastructure. * This shift supports the need for a "Unified Ledger" that can track the entire lifecycle of a transaction—from payment to net profit—across disparate database sources.

netflixOriginal article

Scaling Muse: How Netflix Powers Data-Driven Creative Insights at Trillion-Row Scale | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix’s Muse platform has evolved from a simple dashboard into a high-scale Online Analytical Processing (OLAP) system that processes trillions of rows to provide creative insights for promotional media. To meet growing demands for complex audience affinity analysis and advanced filtering, the engineering team modernized the data serving layer by moving beyond basic batch pipelines. By integrating HyperLogLog sketches for approximate counting and leveraging in-memory precomputed aggregates, the system now delivers low-latency performance and high data accuracy at an immense scale. ### Approximate Counting with HyperLogLog (HLL) Sketches To track metrics like unique impressions and qualified plays without the massive overhead of comparing billions of profile IDs, Muse utilizes the Apache Datasketches library. * The system trades a small margin of error (approximately 0.8% with a logK of 17) for significant gains in processing speed and memory efficiency. * Sketches are built during Druid ingestion using the HLLSketchBuild aggregator with rollup enabled to reduce data volume. * In the Spark ETL process, all-time aggregates are maintained by merging new daily HLL sketches into existing ones using the `hll_union` function. ### Utilizing Hollow for In-Memory Aggregates To reduce the query load on the Druid cluster, Netflix uses Hollow, an internal open-source tool designed for high-density, near-cache data sets. * Muse stores precomputed, all-time aggregates—such as lifetime impressions per asset—within Hollow’s in-memory data structures. * When a user requests "all-time" data, the application retrieves the results from the Hollow cache instead of forcing Druid to scan months or years of historical segments. * This approach significantly lowers latency for the most common queries and frees up Druid resources for more complex, dynamic filtering tasks. ### Optimizing the Druid Data Layer Efficient data retrieval from Druid is critical for supporting the application’s advanced grouping and filtering capabilities. * The team transitioned from hash-based partitioning to range-based partitioning on frequently filtered dimensions like `video_id` to improve data locality and pruning. * Background compaction tasks are utilized to merge small segments into larger ones, reducing metadata overhead and improving scan speeds across the cluster. * Specific tuning was applied to the Druid broker and historical nodes, including adjusting processing threads and buffer sizes to handle the high-concurrency demands of the Muse UI. ### Validation and Data Accuracy Because the move to HLL sketches introduces approximation, the team implemented rigorous validation processes to ensure the data remained actionable. * Internal debugging tools were developed to compare results from the new architecture against the "ground truth" provided by legacy batch systems. * Continuous monitoring ensures that HLL error rates remain within the expected 1–2% range and that data remains consistent across different time grains. For organizations building large-scale OLAP applications, the Muse architecture demonstrates that performance bottlenecks can often be solved by combining approximate data structures with specialized in-memory caches to offload heavy computations from the primary database.