Apache Kylin : Analytical Data Warehouse for Big Data
Page History
Background
Why need a Global Dictionary?
In the field of OLAP data analysis, "count distinct" (or "deduplication") is a very common requirement, and it is divided into approximate deduplication and precise deduplication has the "approximate count distinct" and "precise count distinct" according to the requirements of for the deduplication results.
In large-scale data sets, it is still very challenging to achieve accurate deduplication and ensure quick query response. We know that the most frequently used processing method for precise deduplication is the bit map bitmap method. For integer data, we can save the these integers integer numbers into Bit map Bitmap directly, but in addition to except for integers, there are other data types, such as String. If we want to achieve accurate deduplication, we first need to build a dictionary for these data Perform to do a unified mapping to integers, and then use the Bit map bitmap method for statisticscalculation.
We all know that Kylin uses pre-computation technology to accelerate big data analysis. In the incremental construction of the Cube, in order to avoid errors in the final deduplication results caused by the separate construction of dictionaries for different segments, all segments in a Cube will use the same dictionary (for one column), which is called the global dictionary.
Changes
Kylin has supported the global dictionary function since version 1.5.3, but the construction method at this that time also has obvious defectslimitations:
- The global dictionary is was constructed at a single point on the Job Server, and the construction time becomes uncontrollable as the data increases.
With As the accumulation of data accumulated, the construction of the global dictionary time will require more and more memory for Kylin's construction node. - Limited by the maximum number of integers.
In fact, the Hive-based distributed global dictionary construction has been added to Kylin 3.1. It has solved the above problems. For details, please refer to Kylin Distributed Global Dictionary(Chinese). However, in order to adapt to the new construction query Spark engine, Kylin 4.0 implements another distributed method of building a global dictionary based on sparkSpark. Today we will describe in detail how Kylin 4.0's global dictionary is implementedworks.
Global
...
Dictionary based on Spark
Kylin 4.0 builds a global dictionary based on Spark for distributed encoding processing, which reduces the pressure on a single machine node, and the number of built dictionaries can break the limit of the maximum number of integers.
...
- Every build task will generate a new global dictionary;
- The dictionary of each new build task is saved according to the version number, and the old global dictionary will be gradually deleted;
- A dictionary includes a meta data metadata file and multiple dictionary files, each dictionary file is called a bucket (Bucket);
- Each bucket is divided into two mappings (Map<Object, Long>), and the two are combined into a complete mapping relationship;
draw.io Diagram | ||||||||||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
...
The following figure shows the transfer of the dictionary in the bucket in multiple construction tasks for the bucket numbered 1, and each build creates a new version for the bucket (ie v1, v2, v3, etc.), add the reason behind the version control There will be an explanation. Curr (current) and Prev (Previous) are two HashMaps in a bucket, which store the relative (Relative) code value of the dictionary in the current bucket and the absolute (Absolute) code value of all the dictionary values that have been constructed before.
...
- Create a flat table through Spark and obtain the distinct values that need to be accurately deduplicated;
- Confirm the number of shards according to the number of literal values after deduplication, and determine whether to expand according to demand;
- Repartition data to multiple partitions, encode them separately, and store them in their own dictionary files;
- Assign a version number to the current build task;
- Save dictionary files and metadata data (number of buckets and offset value of buckets);
- According to the conditions, the old version needs to be deleted;
First Build
- Calculate the size of the bucket;
- Take the number of dictionaries that need to be built to handle the maximum value of the single bucket threshold and the default value of the number of buckets.;
- Create buckets and allocate data for encoding;
- Generate the offsets of the meta file metafile record bucket;
The following are related configuration items and their default values.
...
- Determine whether the bucket needs to be expanded according to the number of dictionaries;
- The encoded dictionary value redistributes the expanded bucket;
- Read the latest version of the dictionary data before and distribute it to each bucket;
- Assign the new value to the bucket;
- The value of the dictionary built last time will not change;
...
Version Management
The global dictionary is isolated by assigning a timestamp-based version number to a single build. The reason for adding version control is that the build task may be executed concurrently, and the current coding in the process of building the global dictionary does not support concurrency. Through version control, the global dictionary built before can be completely read for every code, which ensures that the latest version of the dictionary has the most complete global dictionary code, and the global dictionary of a Cube will be read every time it is read Select the latest version of the dictionary. The dictionary is finally stored by version on the file storage system (HDFS here) as shown in the figure below.
...
- Why do I need to use two Maps in a BucketDIctionary?
- At the beginning of the construction process, it is necessary to make a relative code for the dictionary assigned to each bucket starting from 1. The relative code value of this part of the dictionary will be stored in a HashMap. After the relative dictionary value coding is completed, each bucket will be obtained. The offset value, that is, the number of dictionaries in the bucket, and then calculate the absolute (Absolute) code of the dictionary value in each bucket (the bucket is ordered) relative to the offset value of all buckets, and the absolute code of the dictionary is also Will use another HashMap for storage.
- Will there be data skew issues?
- Now that the test is done, the probability that the hotspot cannot be constructed is very small. Generally, it will not pass if it is tilted by one billion level. A lot of columns may indeed cause this problem, but the number of coded buckets can be infinitely enlarged. Unless a single key hotspot is required, adjusting the parameters is also The build is easy to complete.
- Why can the number of global dictionaries exceed the limit of the maximum integer base (2.1 billion)?
- Because of the introduction of the new BitMap data structure Roaring64BitMap, after the global dictionary encoding is completed, the encoding will be compressed into binary and stored in the Roaring64BitMap object. BitMap The bitmap is actually stored through Long instead of Integer.