Kyuubi at Bilibili: Architecture, Enhancements, and Production Practices for Large‑Scale Data Processing

Background

With Bilibili’s rapid business growth, the offline computing cluster expanded from a few hundred nodes to nearly ten thousand across multiple data centers. The platform mainly uses Spark, Presto, and Hive for offline analytics. All BI, ad‑hoc, and data quality services are routed through a self‑developed Dispatcher that selects the optimal engine based on SQL syntax, HDFS read volume, and current engine load. When a query fails, automatic engine downgrade reduces the user barrier. Initially, Spark queries were executed via STS, which suffered performance and reliability issues, prompting the adoption of Kyuubi for multi‑tenant, multi‑engine proxying with full Hive Thrift compatibility.

Query Situation

In ad‑hoc scenarios, Spark SQL accounts for almost half of the queries. Kyuubi’s support for Scala syntax allows advanced users to submit Scala code and switch freely between SQL and Scala modes, greatly enriching ad‑hoc use cases.

Kyuubi Application

Kyuubi is an open‑source project contributed by NetEase to the Apache community. It provides a distributed, multi‑user, JDBC/ODBC‑compatible service for big‑data workloads such as offline analytics, ad‑hoc queries, and BI. It supports popular engines like Spark, Presto, and Flink.

Reasons for choosing Kyuubi:

Full compatibility with the Hive Thrift protocol, matching Bilibili’s existing stack.

High availability and resource isolation essential for large‑scale production.

Extensibility for custom adaptations.

Multi‑engine proxy capability, laying the foundation for a unified compute entry point.

Active community and high‑quality implementation.

Kyuubi Architecture

1. Client : Submits jobs via JDBC or RESTful APIs. 2. Kyuubi Server : Manages sessions, routes them to the appropriate engine. 3. Kyuubi Engine : Executes tasks; different engines have different implementations.

Improvements Based on Kyuubi

3.1 Queue Mode

Kyuubi natively provides CONNECTION, USER, GROUP, and SERVER isolation levels. Bilibili maps departmental YARN queues to GROUP mode and extends it to a custom QUEUE level for finer resource isolation and permission control. Over 20 ad‑hoc queues are managed, each linked to one or more engine instances.

3.2 Multi‑Tenant Support in Queue Mode

The server now starts with the Hive principal, while the Dispatcher submits SQL as a proxy user. Both driver and executor run under the Hive identity, but the proxy user information is propagated so that tasks can be attributed to the correct submitter, enabling accurate accounting and security checks.

3.3 Engine UI Enhancements

Custom Kyuubi listeners capture Spark Job, Stage, Task, and SparkSQL events (SessionCreate, SessionClose, executionStart, executionRunning, executionEnd). These events are aggregated, stored, and displayed on a dedicated Kyuubi UI page, providing real‑time session, job, and SQL statistics.

3.4 Configuration‑Center‑Driven Engine Parameters

Queue‑specific resource parameters (memory, cores) are stored in a central configuration service. When an engine starts, it queries its queue’s settings and overrides the default launch command accordingly.

3.5 Task Progress and Resource Consumption Reporting

The engine collects user, session, job, and stage metrics, runs a periodic job to compute resource costs, and injects the results into operation logs for front‑end visualization.

Stability Enhancements

4.1 Large Result Set Spill to Disk

When driver memory exceeds a threshold during large result fetches, the driver streams results to local disk and serves them back to the client in chunks, preventing OOM and improving stability.

4.2 Task Concurrency Limits

Dynamic limits on the number of concurrent tasks per SQL, per stage (max 300 k tasks), and overall execution time (1 hour timeout) prevent a single large job from monopolizing the engine.

4.3 Table Scan Size Limitation

A custom optimization rule (TableScanLimit) inspects HiveTableRelation and HadoopFsRelation metadata (totalSize, numFiles, numRows) and rejects scans that exceed predefined thresholds.

4.4 Dangerous Join Detection and Join‑Explosion Control

The system identifies potentially hazardous joins (Cartesian, broadcast joins that may cause OOM) and calculates join‑output‑row inflation rates. Queries exceeding safe limits are flagged or rejected.

New Application Scenarios

5.1 Large‑Query Connection & Scala Modes

For heavy or complex ad‑hoc jobs, a dedicated “connection” mode provides an isolated driver and resources, avoiding interference with other users. Users can also mix SQL and Scala code, leveraging the flexibility of Scala while retaining SQL’s ease of use.

5.2 Presto on Spark

To overcome Presto’s memory limits, a Presto‑on‑Spark engine was built on top of Kyuubi. The Presto gateway records query history, decides whether to route a query through Presto‑on‑Spark, obtains a Kyuubi session via ZooKeeper, and submits the query to YARN. Results are fetched back through the gateway.

Deployment Practices

6.1 Kyuubi Server on Kubernetes

Servers are scheduled in a dedicated K8s pool, isolating their environment and enabling rapid scaling.

6.2 Engine on YARN Labels

Engine drivers are placed on a custom YARN node label (kyuubi_label) with enlarged container allocations (<200 GB, 72 vCores), while executors remain on the default label. This separation allows drivers to obtain the resources they need without affecting other workloads.

Future Plans

1. Integrate small ETL tasks into Kyuubi to reduce resource acquisition latency. 2. Make Kyuubi engines (Spark and Flink) cloud‑native and schedule them via K8s. 3. Route Spark‑jar jobs through Kyuubi as well.