Yandex.Metrica 是世界第二大的线上分析系统，Metrica 处理来自网站或者应用的数据流，将它解析成可分析的格式。
Metrica 处理分析消息起来游刃有余，我们主要的技术挑战在于如何以最便于使用的格式来存储处理结果。在整个开发过程中，我们前后几次改变了存储方式。从最初的 MyISAM 表，到后面的 LSM 树，最后变成列式存储数据库 ClickHouse。我将在本文中详解为何我们最后选择自己开发了 ClickHouse。
旧的 Yandex.Metrica 网站有超过40多种固定报表类型（例如：访客地域报告）；几个内页分析工具（例如：点击图）；网站访客（帮助你详细分析单独访客行为）。当然，Yandex.Metrica 还提供单独的报表构造器。
新的 Metrica 和 Appmetrica 系统允许你定制每种报表，而不是以前的“固定”类型查询。你可以增加新的查询维度（例如：搜索词条报告中，你可以增加访问页面进一步分析）、分段、比较（例如：所有的访客数据和旧金山的访客数据比较），改变你的查询指标集等。所以，新的系统依赖的数据存储方式和我们之前使用的完全不一样。
在设计初期，Metrica 被设计为 Yandex.Direct 的分支之一，搜索广告服务。因为 Direct 使用 MySQL 的 MyISAM 引擎，所以 Metrica 开发初期同样沿用了相同的存储。2008 年至 2011 年，MyISAM 引擎被用于存储“固定”的分析报表。
RegionID -> Visits,
SumVisitTime, … 基于这种结构，我们在
WHERE CounterID = AND Date BETWEEN min_date AND max_date 的条件去查询结果。换言之，主键的范围是命中的。
数据在磁盘上如何存储？ How is data actually stored on the disk?
一张 MyISAM 引擎表由一个数据文件和一个索引文件组成。如果更新表的时候不删除数据且不改变长度，数据文件将按照插入的顺序存储每一行序列化的数据。索引（包含主键索引）是 B-树，叶子节点存储数据文件的位移。当我们读取一个索引中的范围数据时，首先从索引中查出一组满足查询条件的数据文件位移，然后按照查出来的位移依次去从数据文件中查找出实际的数据。
Metrica 事件基本按照它们生成的顺序接收处理，在收集端，数据从不同的计数器随机产生。换言之，数据存储时按照时间是连续的，但是按照生产者是不连续的。当写入至一张 MyISAM 表时，来自不同计数器的数据也是非常随机地存储的。这意味着生成报告的时候，需要几行数据，就可能需要执行同样次数的随机检索。
一块经典的 7200 转的硬盘可以每秒钟执行 100 ~ 200 次的随机读取。一个磁盘矩阵，如果使用得当，可以按照比例地执行更多次随机读取。一块使用7年的 SSD 每秒钟可执行 30000 多次的随机读取，但是我们无法支付将数据存放于 SSD 的硬件成本。在目前的系统中，如果我们在一份报表中需要读取10000行数据，大概需要10秒钟，这是完全无法接受的一个时长。
InnoDB 引擎更适合用于主键范围检索，因为它使用聚集主键（聚集索引？）。简单说，数据基于主键以有序的方式存储。但是 InnoDB 的写入速度慢到无法接受。如果你推荐 TokuDB，请继续阅读。
我们采取了一些措施来让 MyISAM 引擎在主键范围检索时更快。
上面（还有一些别的未列举）就是 Yandex.Metrica 使用的优化策略。
尽管 MyISAM 缺点多多，到了 2011 年，我们已经基于它存储了超过 5800 亿的数据。在那之后，所有的数据被转换至 Metrage，因此释放了很多服务器资源。
Metrage and OLAPServer
2010年后我们开始使用 Metrage 存储固定报表，假设你有下述场景：
数据结构中的 LSM 树十分适合上述的业务需求，且比较常见。
A fairly common data structure, LSM Tree, works well for this. This structure consists of a comparatively small group of data "chunks" on the disk, each of which contains data sorted by primary key. New data is initially placed in some type of RAM data structure (MemTable) and then written to the disk in a new, sorted chunk. Periodically a few sorted chunks will be compacted into one larger one in the background. This way a relatively small set of chunks are maintained.
This kind of data structures is used in HBase and Cassandra. Among embedded LSM-Tree data structures, LevelDB and RocksDB are implemented. Subsequently, RocksDB is used in MyRocks, MongoRocks, TiDB, CockroachDB and many others.
Metrage is also an LSM-Tree. Arbitrary data structures (fixed at compile time) can be used as "rows" in it. Every row is a key, value pair. A key is a structure with comparison operations for equality and inequality. The value is an arbitrary structure with operations to update (to add something) and merge (to aggregate or combine with another value). In short, it’s a CRDT.
Both simple structures (integer tuples) and more complex ones (like hash tables for calculating the number of unique visitors or click-map structures) can serve as values. Using the update and merge operations, incremental data aggregation is constantly carried out at the following points:
- during data insertion when forming new batches in RAM
- during background merges
- during read requests
Metrage also contains the domain-specific logic we need that’s performed during queries. For example, for region reports, the key in the table will contain the ID of the lowest region (city, village) and, if we need a country report, the country data will finish aggregating on the database server side.
Here are the main advantages of this data structure:
- Data is located pretty locally on the hard disk; reading the primary key range goes quickly.
- Data is compressed in blocks. Because data is stored in an orderly manner, compression works pretty well when fast compression algorithms are used (in 2010 we used QuickLZ, since 2011 – LZ4).
- Storing data sorted by primary key enables us to use a sparse index. A sparse index is an array of primary key values for each Nth row (N-order of thousands). This index is maximally compact and always fits on the RAM.
Since reading is not performed very often (even though lot of rows are read when it does) the increase in latency due to having many chunks and decompressing the data blocks does not matter. Reading extra rows because of the index sparsity also does not make a difference.
Written chunks of data are not modified. This allows you to read and write without locking – a snapshot of data is taken for reading. Simple and uniform code is used, but we can easily implement all the necessary domain-specific logic.
We had to write Metrage instead of amending an existing solution because there really wasn’t one. LevelDB did not exist in 2010 and TokuDB was proprietary at the time.
All systems that implement LSM-Tree were suitable for storing unstructured data and maps from BLOB to BLOB with slight variations. But to adapt this type of system to work with arbitrary CRDT would have taken much longer than to develop Metrage.
Converting data from MySQL to Metrage was rather time consuming: while it only took about a week for the conversion program to work, the main part of it took about two months to work out.
After transferring reports to Metrage, we immediately saw an increase in Metrica interface speed. We’ve been using Metrage for five years and it has proved to be a reliable solution. During that time, there were only a few minor failures. It’s advantages are its simplicity and effectiveness, which made it a far better choice for storing data than MyISAM.
As of 2015 we stored 3.37 trillion rows in Metrage and used 39 * 2 servers for this. Then we have moved away from storing data in Metrage and deleted most of the tables. The system has its drawbacks; it really only works effectively with fixed reports. Metrage aggregates data and saves aggregated data. But in order to do this, you have to list all the ways in which you want to aggregate data ahead of time. So if we do this in 40 different ways, it means that Metrica will contain 40 types of reports and no more.
To mitigate this we had to keep for a while a separate storage for custom report wizard, called OLAPServer. It is a simple and very limited implementation of a column-oriented database. It supports only one table set in compile time — a session table. Unlike Metrage, data is not updated in real-time, but rather a few times per day. The only data type supported is fixed-length numbers of 1-8 bytes, so it wasn’t suitable for reports with other kinds of data, for example URLs.
Using OLAPServer, we developed an understanding of how well column-oriented DBMS’s handle ad-hoc analytics tasks with non-aggregated data. If you can retrieve any report from non-aggregated data, then it begs the question of whether data even needs to be aggregated in advance, as we did with Metrage database.
On the one hand, pre-aggregating data can reduce the volume of data that is used at the moment when the report page is loading. On the other hand, though, aggregated data doesn’t solve everything. Here are the reasons why:
- you need to have a list of reports that your users need ahead of time
- in other words, the user can’t put together a custom report
- when aggregating a lot of keys, the amount of data is not reduced and aggregation is useless
- when there are a lot of reports, there are too many aggregation options (combinatorial explosion)
- when aggregating high cardinality keys (for example, URLs) the amount of data does not decrease by much (by less than half)
- due to this, the amount of data may not be reduced, but actually grow during aggregation
- users won’t view all the reports that we calculate for them (in other words, a lot of the calculations prove useless)
- it’s difficult to maintain logical consistency when storing a large number of different aggregations
As you can see, if nothing is aggregated and we work with non-aggregated data, then it’s possible that the volume of computations will even be reduced. But only working with non-aggregated data imposes very high demands on the effectiveness of the system that executes the queries.
So if we aggregate the data in advance, then we should do it constantly (in real time), but asynchronously with respect to user queries. We should really just aggregate the data in real time; a large portion of the report being received should consist of prepared data.
If data is not aggregated in advance, all the work has to be done at the moment the user request it (i.e. while they wait for the report page to load). This means that many billions of rows need to be processed in response to the user’s query; the quicker this can be done, the better.
For this you need a good column-oriented DBMS. The market didn’t have any column-oriented DBMS’s that would handle internet-analytics tasks on the scale of Runet (the Russian internet) well enough and would not be prohibitively expensive to license.
Recently, as an alternative to commercial column-oriented DBMS’s, solutions for efficient ad-hoc analytics of data in distributed computing systems began appearing: Cloudera Impala, Spark SQL, Presto, and Apache Drill. Although such systems can work effectively with queries for internal analytical tasks, it is difficult to imagine them as the backend for the web interface of an analytical system accessible to external users.
At Yandex, we developed and later opensourced our own column-oriented DBMS — ClickHouse. Let’s review the basic requirements that we had in mind before we proceeded to development.
Ability to work with large datasets. In current Yandex.Metrica for websites, ClickHouse is used to store all data for reports. As of September, 2017, the database is comprised of 25.1 trillion rows. It’s made up of non-aggregated data that is used to retrieve reports in real-time. Every row in the largest table contains over 500 columns.
The system should scale linearly. ClickHouse allows you to increase the size of cluster by adding new servers as needed. For example, Yandex.Metrica’s main cluster has increased from 60 to 426 servers in three years. In the aim of fault tolerance, our servers are spread across different data centers. ClickHouse can use all hardware resources to process a single query. This way more than 2 terabyte can be processed per second.
High efficiency. We really focus on our database’s high performance. Based on the results of internal tests, ClickHouse processes queries faster than any other system we could acquire. For example, ClickHouse works an average of 2.8-3.4 times faster on web analytics queries than one of top performing commercial column-oriented DBMS (let’s call it DBMS-V).
Functionality should be sufficient for Web analytics tools. The database supports the SQL language dialect, subqueries and JOINs (local and distributed). There are numerous SQL extensions: functions for web analytics, arrays and nested data structures, higher-order functions, aggregate functions for approximate calculations using sketching, etc.
ClickHouse was initially developed by the Yandex.Metrica team. Furthermore, we were able to make the system flexible and extensible enough that it can be successfully used for different tasks. Although the database can run on large clusters, it can be installed on single server or even on a virtual machine.
ClickHouse is well equipped for creating all kinds of analytical tools. Just consider: if the system can handle the challenges of Yandex.Metrica, you can be sure that ClickHouse will cope with other tasks with a lot of performance headroom to spare.
ClickHouse works well as a time series database; at Yandex it is commonly used as the backend for Graphite instead of Ceres/Whisper. This lets us work with more than a trillion metrics on a single server.
ClickHouse is used by analytics for internal tasks. Based on our experience at Yandex, ClickHouse performs at about three orders of magnitude higher than ancient methods of data processing (scripts on MapReduce). But this is not a simple quantitative difference. The fact of the matter is that by having such a high calculation speed, you can afford to employ radically different methods of problem solving.
If an analyst has to make a report and they are competent at their job, they won’t just go ahead and construct one report. Rather, they will start by retrieving dozens of other reports to better understand the nature of the data and test various hypotheses. It is often useful to look at data from different angles in order to posit and check new hypotheses, even if you don’t have a clear goal.
This is only possible if the data analysis speed allows you to conduct instant research. The faster queries are executed, the more hypotheses you can test. Working with ClickHouse, one even gets the sense that they are able to think faster.
In traditional systems, data is like a dead weight, figuratively speaking. You can manipulate it, but it takes a lot of time and is inconvenient. If your data is in ClickHouse though, it is much more malleable: you can study it in different cross-sections and drill down to the individual rows of data.
After one year of open source, ClickHouse is now used by hundreds of companies worldwide. For instance, CloudFlare is using ClickHouse for analytics of DNS traffic, ingesting about 75 billion events each day. Another example is Vertamedia (a video SSP platform), which processes 200 billion events each day in ClickHouse with an ingestion rate of about 3 million rows per second.
Yandex.Metrica has become the second largest web-analytics system in the world. The volume of data that Metrica takes in grew from 200 million events a day in 2009 to more than 25 billion in 2017. In order to provide users with a wide variety of options while still keeping up with the increasing workload, we’ve had to constantly modify our approach to data storage.
Effective hardware utilization is very important to us. In our experience, when you have a large volume of data, it’s better not to worry as much about how well the system scales and instead focus on how effectively each unit of hardware is used: each processor core, disk and SSD, RAM, and network. After all, if your system is already using hundreds of servers, and you have to work ten times more efficiently, it is unlikely that you can just proceed to install thousands of servers, no matter how scalable your system is.
To maximize efficiency, it’s important to customize your solution to meet the needs of specific type of workload. There is no data structure that copes well with completely different scenarios. For example, it’s clear that key-value databases don’t work for analytical queries. The greater the load on the system, the narrower the specialization required. One should not be afraid to use completely different data structures for different tasks.
We were able to set things up so that Yandex.Metrica‘s hardware was relatively inexpensive. This has allowed us to offer the service free of charge to even very large sites and mobile apps, even larger than Yandex’s own, while competitors typically start asking for a paid subscription plan.