Querying Big Data in Real-time with Presto & Grab's TalariaDB

Summary

Grab developed TalariaDB to support real-time SQL querying over high-velocity event streams while maintaining predictable sub-second latencies and low infrastructure costs. The distributed time-series store retains only the most recent hour of data and integrates directly with Presto via its PrestoThriftService interface. Internally, TalariaDB uses the Go-based Badger key-value store to maintain an in-memory key index of metric names and timestamps while mapping columnar event payloads directly to disk. Ingestion occurs by processing pre-partitioned event batches written to Amazon S3 via SQS notifications. By combining a zero-copy decoder with parallel split evaluation across gossiping cluster nodes, the architecture scales horizontally while serving millions of events per second.

Context

Grab needed to query at least 2 to 3 terabytes of high-velocity real-time event data per hour using SQL with predictable low latency and low cost, to enable fast decision-making when experiments impact platform metrics.

Approach / What changed

Grab built TalariaDB, a distributed time-series data store written in Go that ingests columnar event batches from S3 via SQS notifications, stores them in Badger LSM-trees with time-based eviction, and exposes data to Presto via the Presto Thrift Connector.

Takeaways

• TalariaDB uses Badger's key-value store to keep lexicographically ordered metric and timestamp keys in memory while memory-mapping columnar event batches for zero-copy retrieval.
• Ingestion delegates partitioning and batch creation to upstream pipeline services, allowing TalariaDB to ingest files directly from S3 upon receiving SQS notifications.
• Presto queries TalariaDB instances in parallel using the Presto Thrift Connector, while nodes maintain cluster membership via the Gossip protocol and register in Route 53.

Related reading

• Grab · Mar 3, 2019

  How We Simplified Our Data Ingestion & Transformation Process

  Grab evolved its real-time data ingestion pipeline after an initial architecture built on Spark Streaming and Python encountered operational complexity, node failures, and data loss from S3 eventual consistency. Because the streaming workload primarily handled event partitioning and ORC file generation, the team consolidated these tasks directly into an existing Golang processing service. They implemented sharded concurrent maps for high-throughput partitioning and optimized heap allocations to resolve memory bottlenecks. This refactor removed intermediate Avro conversions and intermediate storage hops. The simplified Go pipeline eliminated data loss and reduced processing lag from up to 13 minutes down to approximately 1 minute.

  Yichao Wang

• Grab · May 1, 2017

  Scaling Like a Boss with Presto

  Grab experienced severe performance degradation, long queue times, and connection timeouts on its Amazon Redshift analytics cluster as user concurrency and reporting workloads expanded. Although an initial Amazon S3 data lake decoupled storage from compute, business users required standard SQL interfaces rather than Spark data pipelines. The team deployed Presto clusters on AWS EMR, switching their storage format from AVRO to Parquet to support ANSI SQL querying directly against S3. Utilizing a shared Hive metastore on Amazon RDS allowed Grab to adopt a shared-data multi-cluster architecture that isolated distinct workloads across dedicated compute clusters. This setup enabled rapid cluster scaling, streamlined failover, and matched Redshift performance on partitioned time-range queries.

  Aneesh Chandra

• Grab · Mar 14, 2022

  Real-time data ingestion in Grab

  Service teams at Grab historically had to dual-write transactional data into databases and Kafka, creating data integrity issues during transaction failures alongside substantial schema maintenance overhead. To overcome these limitations and eliminate burst reads from SQL-based queries, the Caspian team built a real-time ingestion platform synchronising MySQL, Aurora, and DynamoDB directly to Kafka. For MySQL and Aurora, the platform uses Debezium with Kafka Connect on ROW-format binlogs, while DynamoDB changes are captured via DynamoDB streams with auto-scaling AWS Lambda functions. Messages encoded in Protobuf are transported via Kafka and ingested into Amazon S3 using a Golang stream processor. This architecture supports search indexing in Elasticsearch, automated data lake pipelines, cross-region disaster recovery replication, and audit trails.

  Shuguang Xiang

• Grab · May 15, 2023

  2.3x faster using the Go plugin to replace Lua virtual machine

  Talaria, an open-source distributed time-series database developed at Grab, previously allowed users to run custom data transformation scripts during ingestion using a Lua virtual machine. Launching and executing Lua scripts caused significant performance overhead when processing large volumes of events. To resolve this bottleneck, the team replaced the Lua VM with Go plugins compiled as Linux shared libraries (.so files). Benchmarks revealed that calling Go plugins achieves performance on par with native Go functions, executing roughly 2.3 times faster and consuming 2.3 times less memory than cached Lua VMs. Both execution methods conform to a unified Handler interface to load and run custom transformations.

  Yonghao Hu