跪拜 Guibai
← Back to the summary

Douyin Group Rebuilds Its Real-Time Data Warehouse on Apache Paimon, Slashing Flink State from Terabytes to Gigabytes

Abstract: This article is based on a presentation by Douyin Group data engineer Su Xing at the Flink Forward Asia 2024 Streaming Lakehouse (I) session. The content is divided into three parts:

  1. Background Introduction
  2. Implementation Practice
  3. Future Outlook

01. Background Introduction

1.1 Business Background

Some platform-based businesses within Douyin Group empower industries with content, primarily using short videos and live streaming as carriers. For gamers, the goal is to provide a complete product chain from "watching games" to "playing games". For gaming influencers, the aim is to inspire them to create diverse gaming content and ensure their creative work receives corresponding rewards. For game developers, a one-stop operational capability solution is provided.

1.2 Data Warehouse Construction

Next, we examine the overall construction of the real-time data warehouse. The real-time data warehouse adopts a layered architecture common in the industry:

1.3 Architecture Implementation

The architecture implementation of the real-time data warehouse is shown in the figure. First, data in the ODS layer mainly comes from client-side tracking, server-side logs, and business database data. The construction from the ODS layer to the APP layer adopts the relatively mature Flink + MQ solution in the industry. In the DWD layer, data ETL and dimension table association (widening) are primarily performed.

Dimension table association is mainly achieved in two ways:

  1. Using KV storage to implement Lookup Join. This method is typically used for scenarios with high timeliness requirements or high traffic. Within Flink, various optimization methods such as Keyed State and distributeBy are used to fully utilize Flink's cache to improve overall query performance. However, under the impact of massive traffic, this solution still poses a significant challenge to the stability of the external KV storage.

  2. Implementing Broadcast Join based on Hive or MySQL. This method is typically used for dimension association for dimension tables with lower timeliness requirements (e.g., T+1 dimension tables). When the Hive partition is ready, it triggers an update of the dimension table.

In the DWD layer, because the internal MQ does not yet support Exactly-Once semantics, data deduplication is required. The APP layer mainly performs customized logic development based on business requirements. Finally, data from the ODS and APP layers is written into downstream OLAP engines or KV storage to provide metric query services externally.

For the entire testing process, taking the DWS layer testing as an example, since MQ does not support direct queries, the MQ data of each layer needs to be synchronized to Hive, and then data comparison is performed based on Hive, resulting in very high overall testing costs. As the business develops, the pain points of the current architecture become increasingly apparent. To solve these problems, the team investigated many open-source data lake engines in the community and ultimately decided to adopt Paimon as the data lake foundation to reconstruct the real-time data warehouse.

1.4 Paimon Introduction

Below is a brief introduction to Paimon's core capabilities. It possesses high-throughput, low-latency data ingestion capabilities, while supporting both streaming subscription and batch queries. Paimon supports mainstream computing engines and OLAP engines, and is particularly tightly integrated with Flink.

1.5 Lakehouse Architecture

Next, let's look at the lakehouse architecture based on Paimon. From the current architecture, it can be seen that compared to the previously introduced Flink + MQ architecture, its design is quite concise. First, in the development process, the DWD layer no longer relies on external KV storage and can directly use Paimon as a dimension table for dimension association via Lookup Join. Moreover, deduplication is no longer needed from the DWD layer to the DWS layer, because the DWS layer can directly consume the Changelog from the DWD layer. This greatly reduces the Keyed State of Flink jobs and ensures data stability. Although the Paimon dimension table is currently stored on HDFS using SSD storage, it still offers significant resource cost benefits compared to the original KV storage.

In terms of development methods, the current model uses Flink Batch for development and debugging, and then converts the job to Flink Streaming mode for online operation when going live. The entire testing process is also different from before. In the new testing process, there is no longer a need to rely on Hive tables, because Paimon supports batch queries and data freshness can reach minute-level, which brings a significant improvement in testing efficiency.

1.6 Architecture Comparison

Next, let's compare the two architectures.

Feature Flink + MQ Architecture Paimon Lakehouse Architecture
Timeliness Second-level latency Minute-level latency (related to Flink Checkpoint interval)
Development Cost Relies on external Hive for development and debugging, high cost Built-in batch read/write capability, no external dependencies, lower cost
Architecture Complexity Needs to integrate multiple external components like KV and Hive, high complexity, high operational cost Converges operations like dimension tables, association, and batch queries internally, concise architecture

02. Implementation Practice

Next, we will introduce the overall implementation practice using two specific real-world scenarios as examples.

2.1 Long-Period Metric Aggregation Calculation Scenario - Background Introduction

First, we examine the calculation in a long-period metric aggregation scenario. After a streamer starts a live broadcast, they want to view the number of user reports for the current live room to optimize their content.

When a user reports a live room, the report information is processed successfully by default. At this time, the information is written to the business database and submitted to the review queue. After the review is completed, the status in the business database is modified. However, the review completion time is not fixed; it could be 30 days or even longer. The real-time data warehouse calculates user report metrics at the live room dimension by consuming the Binlog of the business database, and finally provides external queries through a metric service. In this scenario, the query QPS is around 500.

2.2 Long-Period Metric Aggregation Calculation Scenario - Solution Evolution

The architecture evolution process for this scenario. First, in the Flink + MQ architecture, the message type for this scenario is Changelog. Because MQ does not support transmitting Changelog type data, before metric calculation, it is necessary to manually construct Retract messages using aggregation functions like Last_Value at the user granularity + live room granularity, and then aggregate by live room ID to get the number of user reports for the live room. During this process, the Flink pipeline generates a large amount of Keyed State, and because the review completion time is not fixed, it is impossible to determine how long the state needs to be stored. This causes the state in the Flink task to expand linearly, leading to task instability.

Ultimately, the team adopted a Doris solution, writing user report details directly into Doris's detail table and implementing the overall calculation logic at the query service layer. By performing point queries on the Doris detail table based on the streamer ID and the live room's start/end time, the large state problem in the Flink + MQ pipeline was solved. However, to ensure the overall stability of the Doris cluster, the interface was rate-limited, with the threshold set to 150 QPS. Therefore, when queries encountered high concurrency, rate limiting was frequently triggered, affecting the user's query experience.

2.3 Long-Period Metric Aggregation Calculation Scenario - Paimon Solution

Based on the above problems, this pipeline was reconstructed using Paimon. In the Paimon pipeline, the DWD layer directly creates a Paimon primary key table with user ID, room ID, and date as the primary key, and sets 'changelog-producer' = 'lookup'. In the DWS layer, an aggregation table with live room ID as the primary key is created. In the Flink task, only a relatively simple operation is needed: converting the original boolean type status field to Int type 0 and 1, and writing it directly into the aggregation table. This yields the aggregated metric at the live room granularity. Finally, the calculated metric is written into an external KV store to provide external queries.

2.4 Long-Period Metric Aggregation Calculation Scenario - Business Benefits

The Paimon solution overall solves the large state problem in the Flink solution because it offloads the state computation from Flink to the Paimon storage layer, so the Flink task state is almost negligible. Secondly, because it leverages an external KV store like the Flink solution, its query concurrency can reach millions.

2.5 High-Throughput End-to-End Scenario - Business Background

Next, we examine a high-throughput end-to-end scenario. Internal operations want to view consumption metrics for short video content to adjust their business strategies in a timely manner. However, because the original short video metrics and data do not carry game-related information, it is necessary to ingest the full volume of short video data and associate it with game attributes to calculate aggregated short video metrics in the gaming context. The biggest challenge is that the peak RPS of the full short video data can reach around 8 million, which poses a huge challenge during dimension association.

2.6 High-Throughput End-to-End Scenario - Solution Evolution

Next, we review the overall solution evolution. First, the Flink + MQ approach was adopted. To avoid redundant development, the DWD layer directly ingests the video minute-level metrics from the short video team. In the DWD layer, a Lookup Join is performed directly against an external KV store to obtain game-related information. However, because the peak RPS of the full short video minute-level metrics reaches around 8 million, lookup.cache.ttl was set to 50 minutes in Flink, and Keyed State was enabled to keep the overall cache hit rate above 90%. Despite this, the traffic penetrating to the external KV store could still reach 400,000 per second at peak.

Next, we examine the DWS layer. To avoid multiple associations with the game dimension table, the DWS layer aggregates again based on minute-level metrics to a daily granularity. Because minute-level metrics are constantly changing and their data type is Changelog, but MQ does not support the Changelog type, Last_Value continues to be used here to construct Retract messages within the Flink task. In the short video metric task, a roll-up is performed by video ID and date to ultimately obtain metrics at the short video granularity. However, this process generates a large amount of Keyed State, causing the state to become very large and making the task unstable.

2.7 High-Throughput End-to-End Scenario - Paimon Solution

To solve the above problems, this solution was reconstructed using Paimon. First, the dimension table from the original KV storage was migrated to a Paimon dimension table. Because Paimon's underlying data is stored based on an LSM-Tree structure, it naturally supports point queries in high-throughput, high-concurrency scenarios. And because the data volume of the dimension table averages around 70 million, the Flink Checkpoint interval for the dimension table was set to one minute. Therefore, the data freshness of the current dimension table can basically be within two minutes. For the short video minute-level metrics, a Paimon primary key table is directly created, and the Changelog Producer is set to Lookup mode. In this way, the DWS layer can directly consume the Changelog of the minute-level metrics and perform a roll-up and aggregation directly by short video ID and date, thereby reducing the generation of Keyed State in the Flink task and lowering the state size.

2.8 High-Throughput End-to-End Scenario - Business Benefits

The benefits of the current Paimon solution are as follows:

03. Future Outlook

Finally, let's look at the future outlook.

  1. Explore scalable, low-cost solutions for low-traffic jobs based on Session clusters. Currently, many low-traffic Flink jobs run in Application mode, causing them to occupy resources continuously. The hope is to achieve resource sharing through Flink Session clusters to reduce resource costs.

  2. Explore high-performance synchronization solutions for TB-level dimension tables. A pain point discovered during the use of Paimon is that during cold starts or node migrations, the Paimon dimension table needs to synchronize data from the remote source to the local node, a process that is very slow. Moreover, in certain high-throughput, large dimension table scenarios, the current Paimon solution struggles to solve this problem, so there is a desire to explore high-performance synchronization solutions for TB-level dimension tables through other means.

  3. Continue to utilize capabilities like Partial Update, Tag, and Branch for implementation practice to solve business problems such as multi-stream columnar stitching and data backfilling.