Consistency Architecture for Bilibili Recommendation Model Data Flow

The article describes the recommendation model data flow architecture used at Bilibili, which learns from user behavior to provide personalized video recommendations. It relies on user features, video features from the recommendation service, and user consumption behaviors as training samples.

The data flow consists of two modules: Label Join, which collects user behavior, and Feature Extract, which extracts features from raw logs and computes optimization labels based on user behavior.

In the early architecture, inconsistencies arise: data inconsistency due to timing differences in accessing real-time features stored in Redis, missing features for new videos in offline processing, and feature crossing caused by fixed time windows in Label Join leading to use of latest batch-feature versions during offline extraction.

Calculation inconsistency occurs because feature computation is implemented in three different stacks: online inference (C++), online feature extraction (Flink/Java), and offline feature extraction (Spark/Python), each with different data formats, requiring careful alignment of three separate implementations.

To solve these issues, Bilibili upgraded to a consistency architecture: data consistency is achieved by snapshotting online inference raw features and dumping them to nearline storage, where Label Join and Feature Extract operate on the same snapshot, guaranteeing identical data for training and serving. Calculation consistency is achieved by encapsulating feature computation logic into a C++ library callable via direct invocation in the inference service and via JNI from Java-based offline pipelines, ensuring a single implementation across environments.

The overall architecture involves: the recommendation service packaging raw feature snapshots and sending them via data integration tools to nearline Kafka; nearline Flink latency join performs Label Join, outputting Shitu to Kafka and Hive; nearline Flink computes real-time samples from Shitu, writes to Kafka for online training; offline batch sample computation loads Shitu from Hive via Flink/Spark and writes results back to Hive for batch model training.

Label Join uses Flink latency join with a fixed time window; later enhancements added event-driven triggering based on user behavior patterns (e.g., clear-screen requests) to improve timeliness by 60%, combining event-driven and time-driven strategies.

Sample calculation operates in two modes: online extract (Flink streaming) and offline extract (Flink/Spark batch). It includes stages such as selector, label calculation, in‑request item sampling, and feature generation via a callable pyfe module that invokes a feature library. For recall models, an external candidate pool is used for sampling before feature generation.

BackFill enables algorithm engineers to evaluate new features: for NoDelta mode they read base Shitu directly; for HasDelta mode they join a delta snapshot of new features with baseline Shitu, recompute features, and assess model AUC, iterating if needed. A provided Python SDK supports custom BackFill logic in notebooks or Docker images.

Performance optimization of snapshot processing adopts protobuf wireformat partial decode: instead of deserializing the entire protobuf message (≈7‑8 ms), only required fields are parsed, reducing snapshot handling time from ~14 ms to ~1.5 ms and cutting sample‑calculation CPU usage by over 30 %.

Future work proposes integrating Apache Iceberg to achieve batch‑stream unification: Label Join writes to an Iceberg Shitu table; the sample computation framework reads from Iceberg Shitu and writes to an Iceberg sample table, supporting both real‑time and batch pipelines; training reads from the Iceberg sample table for online or batch training, eliminating duplicate Flink jobs and medium mismatches.

Further optimization leverages Iceberg Merge‑On‑Read (MOR) for incremental BackFill: a baseline sample Iceberg table is branched, new feature columns are added, only delta features are computed and written to the new columns, and the training module merges baseline and incremental features via Iceberg MOR, avoiding full Shitu joins and reducing computational overhead.