Shulou(Shulou.com)06/02 Report--

Today, I will talk to you about how to practice the hot and cold hierarchical architecture design of Wechat backstage massive data. Many people may not understand it very well. In order to make you understand better, the editor summarized the following contents for you. I hope you can get something from this article.

Wechat background data storage with the evolution of Wechat product characteristics, it has undergone several architectural changes to form today's mature large-scale distributed storage system. It methodically manages a machine cluster composed of thousands of heterogeneous machines, which is able to support trillions of access, key values and PB-level data every day.

As a mobile social application based on mobile phone, the data generated by most of the businesses in Wechat have something in common: the data key value is time-stamped, and single-user data is constantly generated over time. We call this kind of data based on time series. For example, the data generated by businesses such as posting in WeChat moments or mobile payment bills and pipelining all meet these characteristics. Data based on time series is naturally hot and cold-this is determined by the physical properties of the phone, and the data displayed on its limited screen can only be divided into screens, through the sliding of fingers. Smooth and continuous access along the timeline-usually from the newly generated data, slowly back to earlier data. At the same time, businesses such as moments are application scenarios for information reading and diffusion, which means that the background data they generate has the distinct characteristics of reading more and writing less.

In the practical application scene of Wechat, the main characteristics of this kind of data include: large amount of data, large number of visits, high degree of importance and so on. These characteristics bring us great challenges in the actual operation of the existing network, mainly including:

Large amount of data, high storage capacity required-data based on time series is usually not deleted, but accumulates over time, the amount of data reaches the PB level, and the corresponding storage space is increasing day by day.

The volume of visits is large, and the holiday effect is obvious-data based on time series are often generated by hot businesses, and their visits remain high, basically at the level of billions of visits per minute. Especially during the festive period, the number of instant visitors can reach three to five times that of the usual.

The importance is high, the user perception is obvious, once the data is lost, the user can not use the product normally, and the complaint rate is high.

Scale-out storage through heap machines can naturally meet the above challenges, but under the premise of tight cost budget, the number of machines is limited. In this case, the hot and cold classification architecture of massive data based on time series arises at the historic moment. The architecture is developed and designed to cope with the increasing expansion of this kind of data in the background, in line with the principle of making full use of machine resources and giving full play to the strengths of various hardware media, combined with the access characteristics of hot and cold data, reading more and writing less. Based on the concept of data layering, it customizes different service strategies according to the differences of data access heat and data volume in different time periods, and expands the storage boundary vertically. Scale-out storage is easy to understand, by adding the same type of machines to the original cluster-which must involve a round of historical data migration-and eventually load balancing between new and old machines, providing services that are not different from each other. Under this scheme, the data flows horizontally and the system is treated equally, obviously there is no place for the idea of adapting measures to local conditions. The architecture of scale-up storage provides such an idea:

For hot spot data, the amount of data is small, but the access traffic is large, so we certainly hope that they can be resident in memory, so the system provides a memory layer with strong consistent guarantee when dealing with sudden traffic. it can also expand the memory layer independently and dynamically without involving historical data migration.

For historical data, the data stock is large, but the access volume is very limited, of course, we do not want to use expensive solid state drives to store them, so the system provides a cheap mechanical disk layer. and there is a set of transparent cold data stripping and batch sinking process, the historical data in the storage layer is continuously extracted to the mechanical disk layer.

Through such an idea of vertical layering and separate expansion, it provides great flexibility for our system, solves the memory bottleneck faced by the storage layer during the holiday season, and alleviates the cost pressure for us from a long-term point of view. the disk capacity bottleneck faced by the storage layer is solved.

Of course, these are not enough for a successful large-scale distributed system, which must also include data multi-replica replication strategy and partition algorithm, as well as the ability to cope with the complex operating environment of the existing network. Combined with the service characteristics of each layer, we formulate the corresponding data strong consistency algorithm, such as the memory layer ensures the complete consistency with the storage layer through version number control, the storage layer realizes multi-copy disaster recovery through Paxos Group, and the mechanical disk layer is guaranteed by serial writing. At the same time, we also implement our own decentralized data routing algorithm, which ensures the uniform distribution of data and traffic, and ensures that this characteristic is still true after horizontal expansion.

Through the above efforts, interlinked, our hot and cold hierarchical architecture of massive data based on time series has successfully responded to the challenges brought by PB-level data, hundreds of billions of access and trillion-level keys.

System design

Data model

The hot and cold hierarchical architecture of massive data mentioned in this paper is dedicated to time series-based data, and their main characteristics are:

a)。 Data key value with timestamp information

b)。 Single-user data is constantly generated over time.

The architecture we designed is strongly dependent on feature a), and each link basically depends on the timestamp in the key value to distribute or sort the data. How to generate the timestamp in the key value, whether to maintain a uniform time globally, and how to maintain it are outside the scope of this article, which is usually determined by the business characteristics of the front end and the time server strategy in the background.

And characteristic b) ensures the necessity and practicability of this architecture. If the size of the data is limited, take the user's account information, for example, it is like the account book in our daily life, it has only one copy, and it will not be added to a single user. Then we usually use a fixed cluster of machines to store it, and there are few changes. What we have to deal with is the user's diary, or bookkeeping book, which is constantly generating new data every day.

We take an example of a cluster in the existing network to illustrate that this kind of business data has the following characteristics:

1. Large amount of data, PB-level data, trillion-level key values, and in the continuous generation, but the newly generated data accounts for a small proportion compared with the historical stock data. The following figure shows a percentage of the cluster data in each time period.

2. The number of visits is large, with a peak of billions of visits per minute, especially during the holiday season, when the high enthusiasm of users can be converted into three to five times the number of visits on weekdays. At the same time, it has the access characteristics of being hot and cold, reading more and writing less (the read-write ratio can even reach 100 per cent). For example, the doubled visits during the festive period are usually visits to the new data generated during the festive period. The following figure shows a percentage of the cluster access in each time period.

3. Data security requirements are high. This kind of data is usually sensitive data perceived by users. Once lost, the user complaint rate is high.

System architecture

The system consists of three layers, as required in the figure, namely, memory layer, storage layer (hot data storage layer) and mechanical disk layer (cold data storage layer). On the timeline, the data they serve ranges from hot to cold. As shown in the following figure:

From the client's point of view, the three layers are juxtaposed, and the client will communicate directly with a machine in a certain layer. The specific difference is that the memory layer and the mechanical disk layer are read-only to the client. All writes are written directly to the storage layer by the client. We distribute the decentralized configuration to the client machine, including memory layer routing, storage layer routing, and other metadata. According to the time separation point and traffic ratio in the configuration, the client decides whether to distribute the current read request to the specific machine of the memory layer or the storage layer. The configuration supports rapid distribution and dynamic loading, and updates can be implemented in seconds.

The routing of the mechanical disk layer is transparent to the client, and the storage layer holds the data link that sinks to the mechanical disk layer, which contains the file number, internal offset, and size, which the client does not know. When the read request for sunken data is distributed to the storage layer machine, the storage layer calculates the machine address of each copy of the data in the cold data storage layer and then replies it to the client together with the file link. According to the random policy, the client chooses a read among multiple replicas. From this point of view, the cold data storage layer is more like a remote file system to the client, while the inode information and routing table is placed in the hot data storage layer.

Next, we will analyze the design strategy of each layer in detail.

Memory layer

The memory layer behaves more like a cache proxy, but ordinary caches are weak in dealing with the validity of data. Common strategies such as elimination when writing, each time before writing to the storage layer, clean up the corresponding data in the cache to ensure invalidation. However, the data is usually multi-copy in the cache, this scheme can not handle network partition errors, and write-out will result in multiple RPC requests, excessive consumption of system resources. Another common strategy is limited data consistency, that is, obsolete obsolescence. When the data is written to the cache, it is accompanied by a valid period of time during which the data is considered to be correct and does not care what the real situation is. This kind of cache can only be applied to services that do not require real-time data. For Wechat's sensitive business, we need distributed caches that can ensure strong consistency of data.

We achieve this through the version number. We maintain a version number for each piece of data in the cache, and there is a corresponding copy in the storage layer. The data in the cache is considered valid only if the version number in the cache matches the version number in the storage tier. Therefore, each read request from the client to the memory layer will be generated by the cache layer and sent to the storage layer. Identify the validity and update the expired data in an RPC request.

Intuitively, strong consistent caching with this scheme does not reduce the access pressure on the storage layer. Because the client request to the cache layer, and the cache layer's request to the storage layer is 1:1. The key to this solution, however, is that we have successfully removed the memory bottleneck in the storage layer. The function of caching data in the storage layer is transferred to the memory of the cache layer. Our requirement for the storage layer is to cache as many version numbers as possible and provide efficient version number access. In this sense, this strongly consistent cache is an extension of storage layer memory. Therefore, we call it the memory layer. Its advantage lies in the dynamic adjustment of traffic ratio and rapid capacity expansion during peak access periods. In later chapters, we also describe how to optimize the resource consumption caused by version number interaction by engineering means.

For the sake of the robustness of the system, some anomalies also need to be considered. If a memory layer machine is suddenly offline, dozens of gigabytes of cached data will fail. Of course, we do not want the pressure of these tens of gigabytes of data. Will all fall on the disk of a storage machine. This will undoubtedly cause the jitter of the system. Therefore, we deployed the memory layer as a group. There are multiple machines in each group. There may be multiple copies of a piece of data in these multiple machines. The client accesses these machines in a random order. In this way, we try our best to avoid the impact of single node failure on the whole system.

In the memory layer, we design a simple and lightweight cache structure that supports variable length data. Each machine contains dozens of LRU chains, each of which is an one-dimensional array in the form of shared memory. All the data is appended to the latest position of the array and loops from the beginning at the end. Naturally, such a structure requires an index to record the location of the data. This approach wastes some memory space, but avoids the dynamic allocation of memory.

Storage layer

The storage layer is at the core of the entire system architecture. Both the memory layer and the machine hard disk layer depend on its implementation. As mentioned earlier, providing efficient and lightweight version number access is the key to the implementation of a strong consistent memory layer. At the same time, the continuous need to sink cold data to the mechanical hard disk layer implies that there must be such a characteristic in the storage layer: cold data is easy to be stripped from all data and collected. This means that if the data in the storage layer is flat and hot and cold data are mixed together, then when we extract the cold data, we have to traverse all the data in the hard disk for a round, which will undoubtedly consume more system resources.

Therefore, we use the lsm-tree algorithm to achieve this requirement. This algorithm is an indexing technique as well as B+ tree. The difference is that it is based on multiple components (components such as C0\ C1) and converts the random IO of the B+ tree into memory operation and sequential IO by delaying commit and merging and sorting. Under our access model, all writes are hot data and will only be submitted to the C0 component. Then, at the right time, merge and sort the data in the C1 component. Through this algorithm, we can achieve the purpose of data layering and data ordering at the same time.

Leveldb is an open source storage engine library of Google, which is based on the idea of lsm-tree algorithm. Therefore, we reuse its mature data structure components, such as log format, data file format, memory table format and so on. However, some of its run-time strategies will bring trouble to our existing network operation. For example, the unrestricted compact policy at runtime and the lazy loading of data file indexes will trigger uncontrollable disk reading and cause service jitter; at the same time, a large number of dynamic memory allocation will also bring some uncontrollable factors to the memory use of the machine. Therefore, we abandoned these runtime behaviors and defined our own management policies to make the system more controllable. At the same time, we make use of the access differences of different data to customize the storage of cold and hot data, define the granularity of block compression and index according to the time period, and effectively reconcile the conversion relationship among system resources such as CPU, memory, disk capacity, disk IO and so on.

The links to cold data and the routing tables of cold clusters are recorded in the storage layer and are not visible to the front end. The specific design ideas are described in detail in the next section.

Mechanical hard disk layer

Although the capacity of mechanical hard disk is large, the performance of IO is low and the failure rate is high. A common idea is that the cold data storage layer is independent of the hot data storage layer, and it is directly visible to the client-the client holds a route of the cold data storage layer and routes it alone-this is undoubtedly a simple and easy-to-understand solution, but there are two problems in our application scenario: inability to accurately defend the air and aggravate the IO tension in the mechanical hard disk layer.

Define the number of TB visits as the number of visits per second of data per TB. In our application scenario, if the actual access volume of each TB historical data is set to N, the service capacity of the mechanical hard disk is only half of that of N. If the cold data storage layer is independent, it needs to maintain all the data indexes on its own, and the memory capacity is not enough to support the index of tens of T of data, so the index can only be dropped from the disk, then each reading of the data will bring about two random reads. Therefore, we put the cold data index and the routing table of the cold data storage layer into the hot data storage layer and are not visible to the front end.

In order to recover from disasters, we must store multiple copies of each data. If the dual copy scheme is adopted, the system needs 50% redundant access capacity to deal with the failure of the other copy, which is not desirable in the case of io bottlenecks. Therefore, we have adopted a three-copy scheme, with only 1/3 redundancy. Each copy is distributed in a different campus, which can achieve disaster recovery at the campus level.

Because of the large capacity and poor computing power of the mechanical disk, we use NO RAID to organize the disk group. In order to better realize the disaster recovery of data loss caused by single disk failure, we achieve exactly the same data at the disk level of three machines in the same group. In order to achieve disk-level load balancing, we achieve monotonous mapping (data-> machine-> disk-> file) by pre-calculating routing and hard coding. The key value of the data can be directly located to the index of the disk and the number of the file.

As a supplement to the mechanical hard disk layer, a cold data sinking module is a must. As the only Writer of the cold data storage layer, we ensure the transparency of the sinking process through two-phase submission, ensure that the use of resources does not affect the existing network service by controlling the timing of process initiation, and ensure that the data is completely consistent at the file level of the cold data storage layer by pre-occupying bits and serial writes.

Strong data consistency guarantee

The business requires that the system must ensure strong consistency among multiple copies of the data. This is an enduring challenge. We will divide the memory layer, the storage layer and the mechanical hard disk layer to consider the strong consistency of the data.

Strong consistent cache

As described earlier, the memory layer, as a strongly consistent distributed cache, is completely aligned to the storage layer, cannot judge the validity of the data, and there is no need for interaction between multiple replicas. It is read-only to the front end, all write requests do not pass through it, and it is only a view of the data in the storage layer. So its commitment to the validity of the front-end data depends entirely on the correctness of the storage layer.

Paxos Group

We achieve the data consistency of the storage layer based on Paxos Group. By using the way of no lease, the system achieves the maximum availability on the premise of strong consistency. The Paxos algorithm is first proposed by Lesile Lamport in this paper, and its only function is to determine a constant value among multiple participants. This is not directly related to distributed storage. On the basis of Paxos algorithm, we design a message-driven Paxos Log component-every operation log is determined by Paxos algorithm, and then realize strong consistency reading and writing based on Paxos Log.

Because Paxos Group uses the ownerless model, all the machines in the group are in the same position at any time. The Paxos algorithm is essentially a multi-copy synchronous writing algorithm, which returns a successful write if and only if the majority in the system accepts the same value. Therefore, the failure of any single node will not cause the unavailability of the system.

The main problem with the strongly consistent writing protocol comes from the Paxos algorithm itself, because multi-stage interaction is needed to ensure that the data is accepted by the majority in the system. We use the following methods to solve the problems in the writing process of the paxos algorithm: based on the fast accept protocol, we optimize the writing algorithm, reduce the number of writes and the number of sending and receiving protocol messages, and finally reduce the writing time and failure; based on random avoidance and limiting the number of times a single Paxos write triggers Prepare, the problem of livelock in Paxos is solved.

The strong consistency read protocol itself does not have much to do with the Paxos algorithm. As long as the majority between multiple replicas is confirmed, the latest data can be obtained. We get the paxos log status of most machines in the cluster (including themselves) by broadcasting, and then judge the validity of the local data.

When the stand-alone node in the system fails and the data is completely lost-this can be regarded as the blind spot of the Paxos algorithm, because the algorithm is based on the fact that all the participants will not break their promise that the data inconsistency caused by the Byzantine failure. -- and this situation is the norm in the current network operation, so we have introduced the Learner Only model. In this mode, the failure machine only receives the submitted data and does not participate in the writing process of the Paxos protocol, which means that no promises will be broken because of data loss. Then quickly recover data from other replicas through asynchronous catch up and full data parity.

In order to prevent multiple nodes from failing at the same time, we distribute multiple copies of data on machines in different campuses. Campus is the concept of different data centers in the same city. In this way, our structure is adequate to deal with a disaster at the level of complete isolation in a single data center.

Serial write

Because it is transparent to the client, the cold data sinking process is the only writer of the mechanical hard disk layer, and the data consistency of that layer is easy to achieve. We achieve data consistency among multiple replicas by writing three copies serially and submitting all of them successfully.

As a supplement, the cold data cluster adds a CRC check and consistency recovery queue to the data block. when the stand-alone data is unavailable (lost or corrupted), the client will first jump to other backups for reading (the three computers provide reading services at the same time), and the consistency recovery queue will recover the local data asynchronously from other backup data blocks.

Because the disk group is organized by No Raid, and the disk-level data files of the machines in the same group are the same, when a single disk failure causes data loss, it is only necessary to transfer the data files from the same sequence disk of other machines.

Data partition

Static mapping table

The main purpose of data partitioning is to ensure load balancing among machines on the same layer, and when the machine size changes, it can still achieve a state of load balancing in the end.

The original intention of the classical consistent hash algorithm is to robust distributed cache and to address based on the dynamic calculation of hash values and virtual nodes at run time. The difference between data storage and distributed cache is that storage must ensure the monotonicity of data mapping, while caching does not require this, so classic consistent hashes usually use machine IP and other parameters for hashing. The result is that the landing point of the data changes from time to time, on the other hand, the load is usually uneven. So we modified the algorithm.

By pre-calculating the random number of virtual nodes, we generate the mapping table between the cutting ring point and the physical machine. The mapping table can support a cluster size of up to 1000 groups, satisfying that in the case of any number of groups, the difference in segment length between entity machines is less than 2%, and the number of arbitrary groups is increased (the upper limit of the total number of groups is no more than 1000 groups). The segment length difference between the changed entity machines is still within 2%. We hard-code this mapping table to avoid the process of calculation at runtime. When the data is addressed according to the hash value of the key, the number of the corresponding entity machine can be obtained after a binary search. We use this mapping table in the memory layer, the storage layer and the mechanical hard disk layer to ensure the consistency of the data routing algorithm in all layers. In terms of engineering implementation, we can reasonably use this feature to batch merge requests to reduce resource consumption, which will be described in detail in a later section.

Intra-group equilibrium

A group is an independent unit of a data partition and an entity unit corresponding to a virtual node. The groups are independent of each other. Each group consists of multiple physical machines, which is the basic unit for Paxos Group to take effect. Multiple copies of one data are scattered on each machine in the group. In order to ensure load balancing on the machines in the group, we also design an algorithm that specifies the access priority between data replicas. The front end will request data one by one according to the priority. As long as the data is successfully obtained, the process will be interrupted. Then we distribute the copies evenly on the machines in the group according to priority, so that the load in the group can be balanced.

Data migration

The static mapping table is very flexible, and one or more groups of machines can be added arbitrarily without reaching the upper limit of the number of groups. Of course, when the route mapping of some data has changed in this process, it involves the movement of historical data. In order not to affect the service in the process of moving, and to ensure that the data is still readable and writable, we have developed a secure and retractable data migration process that is transparent to the front end and based on migration flag bits, a secure and retractable data migration process is realized by means of data double writing and asynchronous data transfer.

Minimum invariant block

The storage layer and the mechanical hard disk layer are coupled together through cold data links. Because the two layers use the same mapping table, when the storage layer is migrated due to capacity expansion, then the cold data link will undoubtedly have to be re-addressed and relocated. If we take the single key value as the granularity to record the cold data link and sink the cold data, then in the context of trillion keys, the efficiency is undoubtedly low. Therefore, we design a minimum invariant block algorithm, through two-stage hashing, using the middle hash bucket to aggregate data, isolating the data key value from the machine routing of the cold data storage layer. Through this algorithm, we can achieve: batch storage of cold data, batch recording of cold data links in blocks (block) in the hot data storage layer, and when the hot data storage layer expands, the data in the block (block) can be migrated to the new target machine as a whole without breaking up due to capacity expansion.

Engineering realization

Poor engineering implementation can destroy a perfect system design, therefore, how to improve the performance of the system by technical means in the process of engineering implementation is also worthy of attention.

Efficient caching

The design of the memory layer relies heavily on the efficient acquisition of the data version number of the storage layer, so it is natural that the version number request is all in memory. Therefore, in view of this situation, we have designed a set of minimalist, lightweight and effective cache for fixed-length version numbers-the memory capacity is not enough to support the full cache of version numbers.

Its data structure is only a two-dimensional array, one-dimensional is used to build hash chain, one-dimensional is used to implement LRU chain. Each read or write needs to be updated by moving the data in the array. In this way, we achieve LRU elimination of the cache as a whole through tens of millions of LRU chain groups. It has the advantages of fixed length, shareable memory, no loss of process restart, high memory utilization and so on.

Batch operation

For the system server, the corresponding logical operations of a request accessed by the front end are serial, so we can naturally comb through the CPU consumption points in this serial process to optimize. However, when the major bottlenecks are gradually eliminated, the CPU consumption points become dispersed and the optimization effect becomes minimal. Therefore, we can only look for other breakthroughs.

We find that there are many logical operation steps in the implementation process of storage engine and consistency protocol algorithm, which involves network interaction, hard disk read and write and so on. Therefore, we decided to merge the same steps in different requests to implement batch operations, which greatly optimized CPU consumption.

The cost of merging is a slight increase in time. Through the separation of speed and slow, we merge only for the logical operations in hot data requests, which removes the instability factor in the time-consuming and reduces the time-consuming jitter.

Request a merge

Since the logical operation performance of the single machine has been greatly improved, the front and back end of the network interaction phase, including the access layer packaging and unpacking, protocol processing and other links, has become the main point of consumption of resources. With reference to the experience of batch operations, we also use batch techniques to optimize performance-that is, a single request accessed from the background (Get) is aggregated into a batch request (Batch Get) at the memory layer.

Routing convergence

Because each data is routed separately according to the key value, if we want to merge requests, we must ensure that the data within the same batch request is addressed to the same Paxos Group. Therefore, we have to aggregate Get requests that fall on the same storage machine at the memory layer. We first use the same routing algorithm in the memory layer and the storage layer, and then align the number of groups in the memory layer with the number of groups in the storage layer to achieve this goal.

Related work

In the design stage, we have fully investigated all kinds of schemes in the industry, ranging from the overall structure of the system to specific technical points. Various solutions have their own application scenarios, each has its own advantages, we can not simply distinguish between good and bad, we are also based on our own business scenarios, carefully choose the right solution, or abandon it. Try to describe it here.

To deal with the data generated by SNS business, there are a variety of hot and cold separation architectures in the industry for reference. We take Facebook's Cold Storage system as an example, it is also based on the idea of hot and cold layering, designed to serve their photo business data storage scheme. The difference is that it adopts the method of combining software and hardware, on the one hand, customizing special servers (including hard disk, power supply, etc.) and data center, on the other hand, reducing the backup number of cold data, increasing erasure codes and other means.

However, we cannot completely apply their experience for two main reasons: the machine models we can use are fixed, and there is no condition for customizing the hardware ourselves. At the same time, it deals with the data of large value such as photos. And we are basically text this type of small value data. From the point of view of the TB traffic mentioned earlier, they are dealing with capacity bottlenecks, while we are dealing with IO bottlenecks, which can be regarded as the challenge of cold data that is not too cold. Therefore, we can only implement our own cold data management strategy.

Similarly, there are many solutions in the industry on how to achieve data consistency. Including our Wechat self-developed Quorum protocol, which is a NWR protocol, using asynchronous synchronous way to achieve multiple copies of data. Even if it is asynchronous synchronization, if multiple copies reach the final consistency, there must be a time difference, so in the case of offline on a single machine, there will be a certain probability that the data will be unavailable. What we pursue is to ensure strong availability of all data in the case of a single point of failure.

Therefore, we use an ownerless decentralized Paxos Group to achieve this goal, in which non-lease is an innovative highlight of the PaxosStore architecture. In case of failure, the system usually wobbles and will be intermittent. In this scenario, the common lease practice is easy to switch hosts repeatedly, resulting in long-term unavailability, while PaxosStore's non-lease structure can easily deal with it and always provide good service. PaxosStore core code is in the process of sorting out open source and is expected to be officially released in the fourth quarter, while the underlying framework of the project is also based on our open source protocol library, github.com/libco.

After reading the above, do you have any further understanding of how to practice the hot and cold hierarchical architecture design of Wechat background massive data? If you want to know more knowledge or related content, please follow the industry information channel, thank you for your support.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.