About SF Technology

Founded in 2009 as an affiliate of SF Express, SF Technology works on building the "brain" for logistics operations and smart logistics services. It has built an integrated big data ecosystem for data collection, synchronization, storage, integration, analysis, mining, visualization and machine learning. Using big data, blockchain, IoT, AI and infrastructure, SF Technology has grown its business in express service, warehousing, cold chain logistics, pharmaceuticals, commerce, finance and other fields.

Running Elasticsearch, ClickHouse, Presto and Apache Kylin side by side made upgrades, engine selection and operations hard. SF Technology introduced StarRocks to provide integrated and consistent big data analytics services.

Key takeaway · Too many engines to run

The challenge

The previous OLAP platform

SF Technology primarily used Elasticsearch, ClickHouse, Presto and Apache Kylin in its OLAP systems:

Elasticsearch is the most widely used OLAP database at SF Technology. Its inverted indexes give it high efficiency and simple operation and maintenance. It is mostly used for log analysis and conditional retrieval. The most used version is Elasticsearch 5.4; newer business uses Elasticsearch 7.6 with customized functions such as cross-region backups, deployment in K8S containers and a data service platform.
ClickHouse was introduced in the past two years and is used in some important scenarios, such as delivery services. It runs in K8S containers to simplify development, deployment and management.
Presto is frequently used to query data in Apache Hive. Presto is deployed on YARN, and the YARN settings are tuned for better integration with Presto while making full use of YARN queues.
Apache Kylin is used less, so far only in a pilot project for the financial department.

Pain points

Container orchestration and customized data analytics platforms had already solved some common problems. Three remained:

OLAP products were hard to upgrade because versions are not compatible with each other. For historical reasons, multiple versions of OLAP products run online at the same time. Users are unwilling to upgrade as long as their business still runs, so upgrades never get done.

Users kept using existing OLAP products in new scenarios they did not suit. Users with a limited understanding of OLAP products picked engines that were wrong for the job: Elasticsearch for aggregating massive amounts of data, Presto for dashboards, Apache Kafka for batch data.

Operating and maintaining the products was difficult. Each OLAP product has its own operation and maintenance methods, and users and operators were expected to know all of them.

Three panels listing pain points of the previous engines. Elasticsearch: poor compatibility across versions, time-consuming upgrades, low cost-performance ratio, unsuited to some OLAP scenarios. ClickHouse: no standard SQL syntax, poor concurrent and join query performance, no automatic data rebalancing, ZooKeeper pressure from concurrent ingestion, average stability, high maintenance cost, hard to compile. Presto: YARN deployment hard to upgrade, weak resource isolation and multi-tenant management, resource utilization in small clusters, choosing between Presto and Trino — Pain points with Elasticsearch, ClickHouse and Presto

Delivery analytics on a single Oracle server

Analytics on delivery-services data was previously done by Oracle on a single machine. It had to handle frequent data updates and provide real-time analytics. As business data grew quickly, the database was overstretched, with problems in availability, analytics speed, data redundancy and timeliness.

Requirements and evaluation

SF Technology summarized its requirements for an OLAP product:

Strong capabilities without obvious weaknesses.
High-quality service, such as stability, anomaly detection and solutions.
Quick response to requests and to major production problems.
Flexibility and room to grow.
Low maintenance cost.

Against these requirements, SF Technology measured several mainstream OLAP products:

Benchmark performance of OLAP databases such as ClickHouse, Presto, Apache Doris and StarRocks.
Join query performance on over one billion rows in medium-sized business scenarios.
SQL queries against ten billion rows of data for delivery services.
Other functions: massive data ingestion, joins on large tables and resilience to failover.

From the evaluation, SF Technology was satisfied with StarRocks in three areas: stability; professional technical support and quick response to requests and problems; and a user-friendly, capable maintenance system.

The solution

StarRocks as one analytics layer

SF Technology introduced StarRocks to provide integrated and consistent big data analytics services. StarRocks ingests data from three kinds of sources:

Real-time and offline data through StarRocks' native loading methods: Stream Load, Broker Load and Routine Load.
ETL data from Apache Flink and Apache Spark through the Apache Flink and Apache Spark connectors.
Data in Apache Hadoop, Elasticsearch and MySQL through external tables.

StarRocks then serves results over JDBC to big data engineering and platform maintenance; Fengjingtai, an interactive BI tool developed by SF Technology; data middle platforms such as data services and data dictionaries; and other business systems that query StarRocks.

In certain scenarios StarRocks takes the place of the other OLAP databases:

Instead of Presto, StarRocks queries Apache Hive data in response to requests from BI tools.
Instead of Elasticsearch, ClickHouse and Apache Kylin, StarRocks stores detailed and aggregated data.

StarRocks also exports data for further processing.

Rollout

Delivery-services data has been loaded into StarRocks as a dark launch. Data from finance, express and international commerce has also been loaded, and more business data will follow.
StarRocks is connected to the BI tools used by SF Technology.
Separate StarRocks clusters serve different purposes: testing before release, shared clusters for production or disaster recovery, and private clusters for important business.
Integration with DataX, Apache Flink and Apache Spark is being developed and tested. StarRocks and SF Technology are also looking at ways to collect detailed data to serve data services and data dictionaries on the middle platforms.

Delivery services on StarRocks

The delivery-services workload had these requirements:

At minimum, support queries against flat tables with about 2,000 columns per row and 80,000 TPS.
Rapid iteration to improve performance.
Store around 10 billion rows of data for the last three months.
Support the transition of the data used to generate 2,000 sets of reports on BI platforms.
Export data for further processing.

Data ingestion

In SF Technology's experience, data processing is most efficient when the number of partitions in the Apache Kafka cluster equals the number of BE nodes.
Ingestion performance depends heavily on row size and data aggregation, so each flat table is split in two: a shared table for less frequently updated data and a private table for frequently updated data.
To handle frequent write requests, the number of drives per machine was increased from 6 to 12, with SSDs under consideration. To work with StarRocks' vectorized engine and reach high QPS, CPU cores were upgraded from 40 to 80.
StarRocks rebalances replicas when a machine breaks down. SF Technology would also like it to rebalance replicas when disks are added or removed.
Problems such as slow ingestion caused by too many ingestion tasks of different versions running in parallel can be handled by adjusting Apache Kafka checkpoints and the number of shards, partitions and replicas.

Queries and availability

StarRocks improved parallel query performance and data caching to keep the BI platform connection stable.
Columns most frequently used for searching and lookups, such as customers' companies, are defined as key columns.
Joined tables are ordered appropriately, especially when the CBO is not running.
When joining two tables, index condition pushdown and redundant fields reduce the volume of data scanned internally.

Data from multiple business systems is processed by Apache Flink and written into a new Apache Kafka cluster. StarRocks extracts it with Routine Load, which gives Exactly-Once semantics, low coupling and high availability.

For higher availability, SF Technology uses dual IDC, double write and an active-active architecture. BI tools and other business applications use two types of domain names for load balancing. With this domain name strategy and JDBC connection load balancing, applications get higher availability, data migration without interruption and resilience to failures.

Two CREATE TABLE statements side by side: the shared table waybill_public for less frequently updated data and the private table pcss_waybill_private for frequently updated data, with callouts 1–8 pointing to the aggregate key, REPLACE_IF_NOT_NULL columns, range partitioning by date, hash bucketing, colocate_with, bloom_filter_columns and dynamic partition properties — Shared and private table design for delivery data

The table designs that improve query performance, numbered as in the figure:

Aggregate model, duplicate model and materialized views.
More parallel ingestion tasks, by splitting one table into two according to update frequency.
Partitioning and bucketing. For example, tables are partitioned by time (a day, a week) and bucketed by the hash of the order column.
Because updates sometimes involve only some columns, REPLACE_IF_NOT_NULL replaces the original value only if the new value is not NULL.
Less frequently updated fields are defined as key columns.
Colocate Join for efficiency.
Dynamic partitioning by date, such as a year or a week, with expired data removed.
Bloom filter indexes for query efficiency.

StarRocksApache KafkaApache FlinkApache SparkApache HiveApache HadoopElasticsearchMySQLFengjingtai

The results

After the data was imported from the previous database systems, StarRocks answers BI tool requests for the 2,000 sets of reports. StarRocks' high availability and autoscaling helped SF Technology handle the business pressure of the Double 11 shopping festival.

StarRocks performs especially well in scenarios such as analyzing less frequently updated data and storing 1 billion rows of data, so SF Technology also uses it outside delivery services, for example in the financial system. The financial system previously often warned of insufficient resources. After deploying StarRocks, it functions well with only 1/3 of the resources.

Three other StarRocks workloads with requirements and solutions. Express delivery incentive mechanism: 3 years of tables, 15 days updated, 50,000 rows a day, maximum 50 QPS, five materialized views. Financial analysis systems: 260 million rows in three tables, 1.5 million rows a month, 1,000 queries a day, daily partitions loaded from Hive. Real-time truck data analytics: six tables kept 15 days, 4 million rows a day, 600 TPS and 200 QPS, bloom filters and local joins — Other StarRocks workloads at SF Technology

What's next

ClickHouse will not provide analytics for SF Technology's newly developed businesses.
In 2022, SF Technology plans to deploy StarRocks at large scale; delivery services already use it.
SF Technology hopes to use StarRocks as its data warehouse in the cloud.

SF Technology plans to build an integrated OLAP analysis platform for all scenarios on StarRocks. For end users, it is intended as a one-stop development and operations platform with:

Serverless, measurable resource management.
High availability in operation and maintenance.
Data modeling for more application scenarios.
Support from multiple database engines for an integrated data analytics platform.
Integration with other applications in the data ecosystem.