About Trip.com
Trip.com is a travel platform that provides booking services for over 1.5 million hotels worldwide. Our big data business intelligence platform, HData, provides data visualization for the company's lodging business line. It presents data through charts so that teams can make decisions quickly and correctly.
ClickHouse could not support high-concurrency queries. Caching and routing low-permission users to MySQL were a temporary fix that left two sets of data and two sets of code to maintain.
Key takeaway · Concurrency was the limit
This article was written by the Trip.com engineering team.
Each team in the lodging department cares about different metrics and has different permissions, so data is presented in many different ways. The site receives about 2,200 UVs and 100,000 PVs every day. The number doubles or even triples during holidays.
The challenge
ClickHouse under concurrent load
We started using ClickHouse in 2018, and 90% of our business line has relied heavily on it since then. ClickHouse kept 95% of response times within 1 s.
We now have about 70 billion records and more than 2,000 processes every day. About 15 billion records need to be updated daily. Uncompressed data is 8 TB; compressed data is 1.75 TB.
However, ClickHouse could not support high-concurrency queries. In most cases CPU consumption is below 30%, but it climbs to a very high level when user queries peak. A single complex, resource-heavy query can fully load the 40-core CPU in a very short time.
Caching and traffic splitting
Visits peak at 9:00 AM on weekdays. To keep the system stable, we reduced the load on the ClickHouse servers with both active and passive caching. We cache the query results of frequently visited pages. After offline data is updated, we also analyze user behavior and cache the relevant data for users who visited it in the last 5 days. This took over a large share of the queries that used to hit ClickHouse, flattened the peaks of concurrent queries and kept us from adding servers without limit.
Real-time data remained a problem, and it matters most during holidays. During the 2021 New Year holiday, real-time dashboard visits were 10 times higher than usual. On weekdays CPU usage stays under 30%; during the holiday it went above 70%, which put the stability of the servers at risk.
We throttled high-frequency user queries on the front end and added a cache on the backend. But real-time data cannot be cached for long: 1–2 minutes is the most users accept. The hit rate of the offline cache is usually 50% or higher, while the real-time cache reaches only 10%.
We also split traffic on the server. Requests from users with fewer permissions need less data, so based on data analysis we set a threshold: queries from users whose permission number is below 5,000 are served by MySQL instead of ClickHouse, which returns their results faster. Users with more permissions still query ClickHouse.
This fixed the immediate problems but created new ones:
- Data consistency between ClickHouse and MySQL is hard to guarantee.
- Hardware cost increases because there are two sets of data.
- ClickHouse does not support standard SQL syntax, so we have to maintain two different sets of code, which raises maintenance cost.
Choosing a new OLAP engine
We needed a new OLAP solution that would reduce development and maintenance cost, keep query performance, and handle high-concurrency, high-throughput scenarios. We tried engines including Ignite, CrateDB and Apache Kylin. Each had advantages in hardware cost or performance, but considering all aspects, StarRocks was our best choice. We chose it mainly for:
- Sub-second latency.
- Performance in complex multi-dimensional analysis, such as high-concurrency and multi-table queries.
- Elastic scaling. Online business is not affected while servers are added, and the backends rebalance data automatically.
- Hot backup for services in the cluster and multi-instance deployment. Node downtime, decommissioning or failures do not affect the overall stability of the cluster.
- Support for materialized views and online schema change.
- Compatibility with the MySQL protocol and standard SQL syntax.
The solution
Performance test against ClickHouse
Queries on HData mostly span multiple tables. ClickHouse usually performs better on a single node than as a cluster, so we tested ClickHouse on a single node and compared it with StarRocks in three test cases.
We set up a StarRocks cluster on 6 virtual machines: 3 with mixed frontend and backend deployment, and 3 for backends only. Version: StarRocks v1.16.2.
ClickHouse version 20.8 ran on the following server:
The test tables:
| Table name | Rows (thousands) |
|---|---|
| overseahotelbasicinfo | 65,000 |
| realtimedailyorder | 1,000 |
| realtimedailyorder_lastweek | 2,000 |
| hotelbasicinfo | 75,000 |
| holdroom_monitor_hotel_value | 3,000 |
| cii_hotel_eidhotel | 45,000 |
| tensity_masterhotel | 10,000 |
Test case 1
- StarRocks: 547 ms
- ClickHouse: 1,814 ms
Test case 2
Test case 3
- StarRocks: 387 ms
- ClickHouse: 884 ms
In these tests, StarRocks query performance was better than ClickHouse.
Data ingestion and updates
HData has two main types of data: real-time data and T+1 offline data. We import real-time data with Routine Load, mainly relying on the update model. T+1 offline data is imported with Stream Load into the duplicate key model. For T+1 data, ETL and import run on Zeus, a data synchronization platform we developed.
Real-time import runs through QMQ, a message queue developed at Trip.com. The real-time import process before StarRocks:
After we integrated StarRocks, the process became:
Keeping order status correct
One scenario subscribes to order status change messages. We use the order ID as the primary key downstream and merge the data with the StarRocks update model, and only non-cancelled orders are shown externally. After receiving a status message, we also add fields by calling external APIs. Calling the interface once per message would put too much pressure on it, so we do this in batches.
Apache Kafka only guarantees message order within a partition, not globally. If two records for the same order with different statuses land in two different partitions, Routine Load cannot know which one will be processed first. If the "canceled" record is processed first, the order wrongly ends up "active" after the other record is processed, which causes errors in later analysis.
We considered using native Apache Kafka instead of QMQ and using the order ID as the key that decides the partition, but that required extra development at a high cost. We chose a compromise instead. While messages are merged, we also write to a log table in the StarRocks duplicate key model that holds only order IDs, order status and message timestamps. A timer task sorts the records with the same order ID, keeps the one with the latest timestamp, matches the order IDs against the original table and updates the status if necessary. This makes sure that the records from Apache Kafka always end up with the correct status.
Disaster recovery and high availability
Our disaster recovery (DR) requirements are high, and the company runs DR exercises regularly. StarRocks has its own disaster recovery mechanism, and we built a high-availability setup on top of it:
- Servers are deployed in 2 server rooms and serve traffic split 5:5. Load balancing for the externally facing frontend nodes is set through configuration items that can be changed and take effect in real time.
- All frontend and backend processes are guarded by a supervisor, so they are restarted automatically if they exit unexpectedly.
- When a frontend node fails, the surviving followers immediately elect a new leader, but applications do not learn of it right away. A timer task checks the frontend servers regularly; when it detects a failed frontend node, it removes the node immediately and sends an SMS to our developers.
- When a backend node fails, StarRocks rebalances the replicas automatically without affecting service. A timer task also checks the backend nodes and emails our developers when one fails.
- We also configured alarms on the hardware indicators of each server. If CPU, memory or disk space looks abnormal, developers are notified and start fixing the issue right away.
The results
At the moment, 70% of the real-time data scenarios on our platform are deployed on StarRocks. The average query response time is about 200 ms, and only 1% of queries take more than 500 ms. We only need to maintain one set of data and one set of code, and the cost of human resources and hardware has decreased significantly.
What's next
- Migrate the remaining real-time scenarios to StarRocks.
- Migrate the offline scenarios to StarRocks as well, and gradually unify all OLAP analysis scenarios on StarRocks.
- Improve monitoring of StarRocks to make it more robust.
- Separate hot and cold data by reading Apache Hive external tables, reducing hardware cost.