How Vipshop’s Hera Data Service Boosts Big Data Access and Performance

Background

Data services are a critical component of a data‑middle‑platform, acting as a unified entry point to the data warehouse. Since 2019, Vipshop has built its own Hera data service to serve more than 30 business lines for both B‑to‑B and B‑to‑C scenarios.

Problems Before a Unified Service

HiveServer‑based daily exports for advertising audience segmentation required 30 minutes + per group, with some jobs exceeding an hour, causing severe latency under resource pressure.

Each data product needed a separate API per storage engine (Presto, ClickHouse, etc.), leading to API explosion and maintenance overhead.

Inconsistent metric definitions across products made data sharing impossible and caused contradictory KPI values.

Solution Overview

Hera provides a single API that abstracts underlying storage and compute engines, supports adaptive SQL generation, unified caching, and ID‑based data product access, thereby improving data delivery efficiency and SLA.

Architecture Design

The system is divided into three layers:

Application Access Layer: Business can connect via TCP client, HTTP, or internal OSP RPC.

Data Service Layer: Handles routing, multi‑engine support, engine resource configuration, dynamic parameter assembly, SQLLispEngine generation, adaptive execution, unified query cache, and FreeMarker‑based SQL construction.

Data Layer: Provides transparent access to data stored in ClickHouse, MySQL, Redis, etc., using the same API.

Scheduling System

Key components:

Master: Manages all Workers, TransferServers, and AdhocWorkers; dispatches jobs.

Worker: Executes ETL and file‑export jobs; launches AdhocWorker processes for ad‑hoc tasks.

Client: Programmatically submits SQL jobs.

ConfigCenter: Pushes unified configuration and SQLParser rules to the cluster.

TransferServer: Handles file transfer services.

Task Scheduling Process

Hera uses a multi‑queue strategy where each queue has its own weight and resource limits. Jobs are scored using:

queue_dynamic_factor = queue_size / queue_capacity * (1 - running_jobs / queue_parallelism)

job_weight = 1 - (current_time - enqueue_time) / timeout

score = job_weight + queue_dynamic_factor + queue_weight

Higher scores increase a job’s chance of being selected when resources are limited.

Core Features

Multi‑queue scheduling : Separate queues for different users and task types, respecting SLA.

Multi‑engine query : Supports Spark, Presto, ClickHouse, Hive, MySQL, Redis, automatically picking the best engine.

Multiple task types : ETL, ad‑hoc, file export, data import.

File export : SQL → distributed engine → HDFS/Alluxio → TCP download; large‑scale exports reduced from >30 min to ≤3 min (10‑30× speedup).

Resource isolation : Physical isolation of worker and engine resources per business.

Dynamic engine parameter assembly : Engine‑specific parameters are generated and can be adjusted at runtime.

Adaptive engine execution : If the chosen engine fails, Hera automatically falls back to another engine (see diagram).

SQL construction via Lisp DSL : Custom Lisp syntax describes metric calculations, case expressions, type casts, and generic functions. Example:

(count x [y,z...])

(case (= subject_id (string goods_base)) (int 2) (int 1))

(cast bigint xx)

These expressions are parsed with ANTLR4, transformed into engine‑specific SQL, and executed.

Metrics Collection

Hera collects static metrics (node info) and dynamic metrics (runtime stats). Workers send heartbeat messages containing memory usage and task counts, enabling the Master to make informed scheduling decisions.

Production Usage

Daily call volume: >900 M to‑C requests, >1.5 M to‑B requests.

ETL jobs typically finish in ~3 minutes.

Ad‑hoc queries: 90 % complete within 2 seconds; ClickHouse queries 99 % within 1 second; Presto ~50 W+ calls/day, ClickHouse ~44 W+ calls/day.

Performance Improvements for Crowd Computation

To address SLA issues in audience segmentation, Hera co‑locates compute and storage, replaces HDFS tables with Alluxio cache tables, and synchronizes partitions via a scheduled task. Example Hive table creation:

CREATE TABLE `hdfs.ads_tags_table`(
  `oaid_md5` string,
  `mid` string,
  `user_id` bigint,
  ...
) PARTITIONED BY (`dt` string) LOCATION 'hdfs://xxxx/hdfs.db/ads_tags_table';

Alluxio counterpart:

CREATE TABLE `alluxio.ads_tags_table`(
  `oaid_md5` string,
  `mid` string,
  `user_id` bigint,
  ...
) PARTITIONED BY (`dt` string COMMENT '????') LOCATION 'alluxio://zk@IP1:2181,IP2:2181/alluxio.db/ads_tags_table';

Synchronization workflow: a timed task detects new partitions, triggers a SYN2ALLUXIO job, which adds matching partitions to the Alluxio table; Alluxio then pulls data from the mounted HDFS path.

Sample Spark ETL SQL (original):

INSERT INTO hive_advert.cd0000127760_full
SELECT result_id, '20210703'
FROM (
  SELECT oaid_md5 AS result_id
  FROM hdfs.ads_tags_table AS ta
  WHERE ta.dt = '20210702' AND ...
) AS t;

After rewrite to use the Alluxio cache table:

INSERT INTO hive_advert.cd0000127760_full
SELECT result_id, '20210703'
FROM (
  SELECT oaid_md5 AS result_id
  FROM alluxio.ads_tags_table AS ta
  WHERE ta.dt = '20210702' AND ...
) AS t;

Using Alluxio and Spark mixed‑deployment, crowd‑computation jobs achieve a 10‑30 % speedup.

Remaining Challenges

Inconsistent function signatures across engines (e.g., Presto strpos vs. ClickHouse position) require a more elegant abstraction.

Adoption of ClickHouse bitmap solutions for audience segmentation.

Improving HA, disaster recovery, and K8s deployment for the scheduling subsystem.