- Product Introduction
- Purchase Guide
- Getting Started
- Operation Guide
- Development Guide
- Design of Data Table
- Importing Data
- Import Overview
- Import Methods
- Synchronizing Data with JDBC
- Synchronizing Data Through External Tables
- Real-time or Batch Writing of MySQL Data
- Importing Data from Kafka
- Importing with DataX
- Importing Doris from Logstash
- Data Transformation, Column Mapping and Filtering
- Importing Strict Mode
- Importing Data in JSON Format
- Exporting Data
- Basic Feature
- Query Optimization
- Ecological Expansion Feature
- Product Policy
- Product Introduction
- Purchase Guide
- Getting Started
- Operation Guide
- Development Guide
- Design of Data Table
- Importing Data
- Import Overview
- Import Methods
- Synchronizing Data with JDBC
- Synchronizing Data Through External Tables
- Real-time or Batch Writing of MySQL Data
- Importing Data from Kafka
- Importing with DataX
- Importing Doris from Logstash
- Data Transformation, Column Mapping and Filtering
- Importing Strict Mode
- Importing Data in JSON Format
- Exporting Data
- Basic Feature
- Query Optimization
- Ecological Expansion Feature
- Product Policy
Requirement Scenarios
In most data analysis scenarios, there is more data writing than reading, such as writing once and reading frequently. Take a report involving the dimensions and metrics as an example. The calculation is done at the midnight, but there are hundreds, if not thousands, of page visits daily. It is therefore very suitable to cache the result sets. In data analysis or BI applications, there are the following business scenarios:
High concurrency scenarios. Doris can support high concurrency well, but a single server cannot bear too high QPS.
Complex dashboard. Complex dashboard or big-screen applications, with data from multiple tables and dozens of queries per page. Although each query only lasts tens of milliseconds, the total query time will be a few seconds.
Trend analysis. Queries within a given date range, metrics displayed daily. For example, querying trends in the number of users in the last 7 days. This type of query involves large data volumes, wide query range, and usually takes tens of seconds.
Repeated user queries. If there is no mechanism to prevent repeated refreshes for the product, the user might refresh the page repeatedly due to mistaken operations or other reasons, submitting lots of duplicate SQL.
For the above four scenarios, the solution at the application layer is to put query results in Redis, periodically update the cache, or refresh the cache manually. However, this solution has the following problems:
Data inconsistency. It is unable to perceive data updates, so the users often see outdated data.
Low hit rate. If the entire query result is cached and data is continually written, the cache will frequently expire, resulting in a low hit rate and high system load.
Additional costs. Introducing external cache components will make the system more complicated and increase additional costs.
Solutions
This partition caching strategy can solve the above problems. It primarily ensures data consistency, then refines the caching granularity to enhance the hit rate. Hence, it features:
Users don't need to worry about data consistency. The expiration of the cache is controlled by the version. The cached data and the data queried from BE are the same.
No additional components or costs. The cached result is stored in BE's memory, and users can adjust cache size as needed.
Implementation of two caching strategies: SQLCache and PartitionCache. The latter has finer caching granularity.
Using consistent hashing to solve the online and offline of BE nodes. The cache algorithm in BE is an improved LRU.
SQLCache
SQLCache stores and accesses the cache by the SQL signature, the partition ID of the queried table, and the latest partition version. The cache dataset is determined by these three factors. Any changes, such as changes in the SQL statement, differences in query fields or terms, or version changes after data update, will lead to cache miss.
If multiple tables are joined, the nearest updated partition ID and the latest version number are used. An update in any table will cause the partition ID or version number to differ and also lead to a cache miss.
SQLCache is more suitable for T + 1 update scenarios, where data is updated in the early hours of the morning. The first query accesses the result from the BE and places it into the cache, and all subsequent identical queries access the cache. Real-time data updates can also use it, but there may be issues of a low hit rate. Seethe following PartitionCache for more details.
PartitionCache
Design Principle
1. SQL queries can be split in parallel, where Q = Q1 ∪ Q2 ... ∪ Qn and R = R1 ∪ R2 ... ∪ Rn, Q represents the query statement, and R represents the result set.
2. Split into read-only partitions and updateable partitions. Cache only the read-only partitions and do not cache the updateable partitions.
As shown above, when querying the number of users from the last 7 days, if partitioned by date, data is only written to the current day's partition. Data from partitions other than the current day is fixed and unchanging. Under the same query SQL, the metrics queried from a non-updated partition are fixed. As illustrated below, when querying the number of users for the past 7 days on March 9, 2020, the data from March 3, 2020 to March 7, 2020 is retrieved from the cache. The query for data from March 8, 2020 comes from the partition for the first time, and subsequent queries come from the cache. The data of March 9, 2020 comes from the partition due to continuous write operations on the day.
Therefore, when querying data from the past N days, with the data updated in the most recent D days, similar queries varying only by date range need to query just D partitions. The rest comes from cache, effectively reducing cluster load and query time.
MySQL [(none)]> SELECT eventdate,count(userid) FROM testdb.appevent WHERE eventdate>="2020-03-03" AND eventdate<="2020-03-09" GROUP BY eventdate ORDER BY eventdate;+------------+-----------------+| eventdate | count(userid) |+------------+-----------------+| 2020-03-03 | 15 || 2020-03-04 | 20 || 2020-03-05 | 25 || 2020-03-06 | 30 || 2020-03-07 | 35 || 2020-03-08 | 40 | // Coming from the partition at first, then from the cache| 2020-03-09 | 25 | // Coming from the partition+------------+-----------------+7 rows in set (0.02 sec)
In PartitionCache, the primary Key of the cache is the 128-bit MD5 Signature of the SQL, after stripping the partition condition. The following is the rewritten SQL to be signed:
SELECT eventdate,count(userid) FROM testdb.appevent GROUP BY eventdate ORDER BY eventdate;
The secondary Key is the content of the result set's partition field, such as the content of the eventdate column in the above query result. The Subordinate information of the secondary Key is the partition's version number and version update time.
The following shows the process of executing the above SQL for the first time on March 9, 2020:
1. Access data from the cache.
+------------+-----------------+| 2020-03-03 | 15 || 2020-03-04 | 20 || 2020-03-05 | 25 || 2020-03-06 | 30 || 2020-03-07 | 35 |+------------+-----------------+
2. The SQL and data accessed from BE.
SELECT eventdate,count(userid) FROM testdb.appevent WHERE eventdate>="2020-03-08" AND eventdate<="2020-03-09" GROUP BY eventdate ORDER BY eventdate;+------------+-----------------+| 2020-03-08 | 40 |+------------+-----------------+| 2020-03-09 | 25 |+------------+-----------------+
3. The data sent to the terminal at the end.
+------------+-----------------+| eventdate | count(userid) |+------------+-----------------+| 2020-03-03 | 15 || 2020-03-04 | 20 || 2020-03-05 | 25 || 2020-03-06 | 30 || 2020-03-07 | 35 || 2020-03-08 | 40 || 2020-03-09 | 25 |+------------+-----------------+
4. The data sent to the cache.
+------------+-----------------+| 2020-03-08 | 40 |+------------+-----------------+
Partition caching is suitable for partitions by date, with some partitions updated in real-time, and the SQL query is relatively fixed. The partition field can also be other fields, but it is necessary to ensure that only a small number of partitions are updated.
Limits
It only supports OlapTable. Other storages such as MySQL tables do not have version information, and it is impossible to perceive whether the data has been updated.
Only supports grouping by the partition field, grouping by other fields is not supported. If groups by other fields, the group data may be updated, rendering the cache ineffective.
Only the first half of the result set is supported. The later half and all parts that hit the cache are unsupported if the result set is divided into several parts by the cached data.
Use Method
Enabling SQLCache
1. Ensure the cache_enable_sql_mode=true (the default is true) in fe.conf.
vim fe/conf/fe.confcache_enable_sql_mode=true
2. Setting variables in the MySQL command line.
MySQL [(none)]> set [global] enable_sql_cache=true;
Note
"global" is a global variable, without specifying it indicates the variable of current session.
Enabling PartitionCache
1. Ensure the cache_enable_partition_mode=true (the default is true) in fe.conf.
vim fe/conf/fe.confcache_enable_partition_mode=true
2. Setting variables in the MySQL command line.
MySQL [(none)]> set [global] enable_partition_cache=true;
If both caching strategies are enabled at the same time, pay attention to the following parameters:
cache_last_version_interval_second=900
If the latest version of the partition is further from the current time than cache_last_version_interval_second, the entire query result will be cached. If it is less than this interval and meets the conditions of PartitionCache, then the data will be cached according to PartitionCache.
Monitoring
FE Monitoring Metrics:
query_table //Number of tables in the queryquery_olap_table //Number of Olap tables in the querycache_mode_sql //Number of Queries recognized with cache pattern as sqlcache_hit_sql //Number of Queries with the sql pattern hitting the cachequery_mode_partition //Number of Queries identified with the partition pattern for cachingcache_hit_partition //Number of Queries hitting the partitionpartition_all //All partitions scanned in the Querypartition_hit //Number of partitions hit in CacheCache hit rate = (cache_hit_sql + cache_hit_partition) / query_olap_tablePartition hit rate = partition_hit / partition_all
Monitoring items for BE:
query_cache_memory_total_byte //Cache memory sizequery_query_cache_sql_total_count //Number of SQL in Cachequery_cache_partition_total_count //Number of Cache partitionsAverage SQL data size = cache_memory_total / cache_sql_totalAverage partition data size = cache_memory_total / cache_partition_total
Other Monitoring:
The CPU and memory indicators of the BE node can be checked from Grafana, the Query Percentile and other indicators in Query statistics, and achieve business objectives in conjunction with the adjustment of cache parameters.
Optimizing Parameters
The configuration item cache_result_max_row_count in FE, which is the maximum number of rows that the query result set is placed into Cache and it can be adjusted according to the actual situation. However, it is suggested not to set too large to avoid excessive memory occupancy. The result set exceeding this size will not be cached.
vim fe/conf/fe.confcache_result_max_row_count=3000
The maximum partition number cache_max_partition_count in BE refers to the maximum partition number corresponding to each SQL. If partitioned by date, it can cache more than two years of data. If you want to preserve Cache for a longer time, please set this parameter larger and simultaneously adjust the parameter cache_result_max_row_count.
vim be/conf/be.confcache_max_partition_count=1024
There are two parameter configurations for Cache memory in BE, namely, query_cache_max_size and query_cache_elasticity_size (unit: MB). When the memory exceeds query_cache_max_size + cache_elasticity_size, the system starts cleaning up and controls the memory to below query_cache_max_size. These two parameters can be set according to the number of BE nodes, node memory size, and Cache hit rate.
query_cache_max_size_mb=256query_cache_elasticity_size_mb=128
Calculation Method:
Assuming Cache/Caching 10K queries, each query caches 1000 rows, each row is 128 bytes, distributed on 10 BE nodes, then each BE needs 128M memory (10K1000128/10).
Note
T+1 data does not currently support the use of Partition Cache.
For similar SQL, if you have previously queried two indicators and now query three indicators, the current system does not support the use of Cache for the two indicators.
If partitioned by date, but aggregation of data is required on a weekly basis, PartitionCache currently does not support.
Yes
No
Was this page helpful?