Showing posts with label BigTable. Show all posts
Showing posts with label BigTable. Show all posts

Tuesday, October 12, 2021

LSM tree (log-structured merge-tree) > Memtable | Sparse index

作者:henryPKU
链接:https://www.zhihu.com/question/19887265/answer/1714901833
来源:知乎
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。

LSM tree (log-structured merge-tree) 是一种对频繁写操作非常友好的数据结构,同时兼顾了查询效率。LSM tree 是许多 key-value 型或日志型数据库所依赖的核心数据结构,例如 BigTableHBaseCassandraLevelDBSQLiteScyllaRocksDB 等。

LSM tree 之所以有效是基于以下事实:磁盘或内存的连续读写性能远高于随机读写性能,有时候这种差距可以达到三个数量级之高。这种现象不仅对传统的机械硬盘成立,对 SSD 硬盘也同样成立。如下图:



LSM tree 在工作过程中尽可能避免随机读写,充分发挥了磁盘连续读写的性能优势。

SSTable

LSM tree 持久化到硬盘上之后的结构称为 Sorted Strings Table (SSTable)。顾名思义,SSTable 保存了排序后的数据(实际上是按照 key 排序的 key-value 对)。每个 SSTable 可以包含多个存储数据的文件,称为 segment,每个 segment 内部都是有序的,但不同 segment 之间没有顺序关系。一个 segment 一旦生成便不再修改(immutable)。一个 SSTable 的示例如下:



可以看到,每个 segment 内部的数据都是按照 key 排序的。下面我们来介绍每个 segment 是如何生成的。

写入数据

LSM tree 的所有写操作均为连续写,因此效率非常高。但由于外部数据是无序到来的,如果无脑连续写入到 segment,显然是不能保证顺序的。对此,LSM tree 会在内存中构造一个有序数据结构(称为 memtable),例如红黑树。每条新到达的数据都插入到该红黑树中,从而始终保持数据有序。当写入的数据量达到一定阈值时,将触发红黑树的 flush 操作,把所有排好序的数据一次性写入到硬盘中(该过程为连续写),生成一个新的 segment。而之后红黑树便从零开始下一轮积攒数据的过程。



读取/查询数据

如何从 SSTable 中查询一条特定的数据呢?一个最简单直接的办法是扫描所有的 segment,直到找到所查询的 key 为止。通常应该从最新的 segment 扫描,依次到最老的 segment,这是因为越是最近的数据越可能被用户查询,把最近的数据优先扫描能够提高平均查询速度。

当扫描某个特定的 segment 时,由于该 segment 内部的数据是有序的,因此可以使用二分查找的方式,在 O(logn) 的时间内得到查询结果。但对于二分查找来说,要么一次性把数据全部读入内存,要么在每次二分时都消耗一次磁盘 IO,当 segment 非常大时(这种情况在大数据场景下司空见惯),这两种情况的代价都非常高。一个简单的优化策略是,在内存中维护一个稀疏索引(sparse index,其结构如下图:



稀疏索引 sparse index | 全量索引(dense index)| March 28, 2022 review 

稀疏索引是指将有序数据切分成(固定大小的)块,仅对各个块开头的一条数据做索引。与之相对的是全量索引(dense index),即对全部数据编制索引,其中的任意一条数据发生增删均需要更新索引。两者相比,全量索引的查询效率更高,达到了理论极限值 O(logn),但写入和删除效率更低,因为每次数据增删时均需要因为更新索引而消耗一次 IO 操作。通常的关系型数据库,例如 MySQL 等,其内部采用 B tree 作为索引结构,这便是一种全量索引。

有了稀疏索引之后,可以先在索引表中使用二分查找快速定位某个 key 位于哪一小块数据中,然后仅从磁盘中读取这一块数据即可获得最终查询结果,此时加载的数据量仅仅是整个 segment 的一小部分,因此 IO 代价较小。以上图为例,假设我们要查询 dollar 所对应的 value。首先在稀疏索引表中进行二分查找,定位到 dollar 应该位于 dog downgrade 之间,对应的 offset 17208~19504。之后去磁盘中读取该范围内的全部数据,然后再次进行二分查找即可找到结果,或确定结果不存在。

稀疏索引极大地提高了查询性能,然而有一种极端情况却会造成查询性能骤降:当要查询的结果在 SSTable 中不存在时,我们将不得不依次扫描完所有的 segment,这是最差的一种情况。有一种称为布隆过滤器(bloom filter的数据结构天然适合解决该问题。布隆过滤器是一种空间效率极高的算法,能够快速地检测一条数据是否在数据集中存在。我们只需要在写入每条数据之前先在布隆过滤器中登记一下,在查询时即可断定某条数据是否缺失。

布隆过滤器的内部依赖于哈希算法,当检测某一条数据是否见过时,有一定概率出现假阳性(False Positive),但一定不会出现假阴性(False Negative)。也就是说,当布隆过滤器认为一条数据出现过,那么该条数据很可能出现过;但如果布隆过滤器认为一条数据没出现过,那么该条数据一定没出现过。这种特性刚好与此处的需求相契合,即检验某条数据是否缺失。

文件合并(Compaction

随着数据的不断积累,SSTable 将会产生越来越多的 segment,导致查询时扫描文件的 IO 次数增多,效率降低,因此需要有一种机制来控制 segment 的数量。对此,LSM tree 会定期执行文件合并(compaction)操作,将多个 segment 合并成一个较大的 segment,随后将旧的 segment 清理掉。由于每个 segment 内部的数据都是有序的,合并过程类似于归并排序,效率很高,只需要O(n)的时间复杂度。



在上图的示例中,segment 1 2 中都存在 key dog 的数据,这时应该以最新的 segment 为准,因此合并后的值取 84 而不是 52,这实现了类似于字典/HashMap 覆盖写的语义。

删除数据

现在你已经了解了 LSM tree 读写数据的方式,那么如何删除数据呢?如果是在内存中,删除某块数据通常是将它的引用指向 NULL,那么这块内存就会被回收。但现在的情况是,数据已经存储在硬盘中,要从一个 segment 文件中间抹除一段数据必须要覆写其之后的所有内容,这个成本非常高。LSM tree 所采用的做法是设计一个特殊的标志位,称为 tombstone(墓碑),删除一条数据就是把它的 value 置为墓碑,如下图所示:



这个例子展示了删除 segment 2 中的 dog 之后的效果。注意,此时 segment 1 中仍然保留着 dog 的旧数据,如果我们查询 dog,那么应该返回空,而不是 52。因此,删除操作的本质是覆盖写,而不是清除一条数据,这一点初看起来不太符合常识。墓碑会在 compact 操作中被清理掉,于是置为墓碑的数据在新的 segment 中将不复存在。

LSM tree B tree 的对比

主流的关系型数据库均以 B/B+ tree 作为其构建索引的数据结构,这是因为 B tree 提供了理论上最高的查询效率 - O(logn)。但对查询性能的追求也造成了 B tree 的相应缺点,即每次插入或删除一条数据时,均需要更新索引,从而造成一次磁盘 IO。这种特性决定了 B tree 只适用于频繁读、较少写的场景。如果在频繁写的场景下,将造成大量的磁盘 IO,从而导致性能骤降。这种应用场景在传统的关系型数据库中比较常见。

LSM tree 则避免了频繁写场景下的磁盘 IO 开销,尽管其查询效率无法达到理想的 ,但依然非常快,可以接受。所以从本质上来说,LSM tree 相当于牺牲了一部分查询性能,换取了可观的写入性能。这对于 key-value 型或日志型数据库是非常重要的

总结

LSM tree 存储引擎的工作原理包含以下几个要点:

  1. 写数据时,首先将数据缓存到内存中的一个有序树结构中(称为 memtable)。同时触发相关结构的更新,例如布隆过滤器、稀疏索引。
  2. memtable 积累到足够大时,会一次性写入磁盘中,生成一个内部有序的 segment 文件。该过程为连续写,因此效率极高
  3. 进行查询时,首先检查布隆过滤器。如果布隆过滤器报告数据不存在,则直接返回不存在。否则,按照从新到老的顺序依次查询每个 segment
  4. 在查询每个 segment 时,首先使用二分搜索检索对应的稀疏索引,找到数据所在的 offset 范围。然后读取磁盘上该范围内的数据,再次进行二分查找并获得结果
  5. 对于大量的 segment 文件,定期在后台执行 compaction 操作,将多个文件合并为更大的文件,以保证查询效率不衰减

参考:

 Actionable Items

It takes time to make the article to be readable. I also learn a lot about memtable and SSTable. The example is very helpful for me to understand how BigTable is designed to use LSM similar design. 

Follow up 

March 28, 2022

Google those keywords:

  1. LSM tree 存储引擎的工作原理
  2. 将数据缓存到内存中的一个有序树结构中(称为 memtable
  3.  memtable 积累到足够大时,会一次性写入磁盘中,生成一个内部有序的 segment 文件。
  4. 布隆过滤器、稀疏索引, bloom filter, sparse index 
  5. Segment - 
  6. Binary search - sparse index - offset range - read disk date - binary search and find result
  7. 对于大量的 segment 文件,定期在后台执行 compaction 操作,将多个文件合并为更大的文件,以保证查询效率不衰减
  8.  key-value 型或日志型数据库所依赖的核心数据结构
  9. LSM tree (log-structured merge-tree) 
  10. BigTableHBaseCassandraLevelDBSQLiteScyllaRocksDB
  11. 稀疏索引 sparse index | 全量索引(dense index)| March 28, 2022 review 
  12. Segment compaction | tombstone | How to delete the key? 
  13. LSM tree 的所有写操作均为连续写,因此效率非常高。但由于外部数据是无序到来的,如果无脑连续写入到 segment,显然是不能保证顺序的。对此,LSM tree 会在内存中构造一个有序数据结构(称为 memtable),例如红黑树。每条新到达的数据都插入到该红黑树中,从而始终保持数据有序。当写入的数据量达到一定阈值时,将触发红黑树的 flush 操作,把所有排好序的数据一次性写入到硬盘中(该过程为连续写),生成一个新的 segment。而之后红黑树便从零开始下一轮积攒数据的过程。
  14. LSM tree -> segment -> memtable - using red black tree -> Write once for a block into hard disk/ SSDisk
  15. Write -> a lot of write -> no order -> memtable -> SStable -> segment -> GFS data structure -> BigTable data structure 


Tuesday, August 17, 2021

Dean Keynote Ladis 2009: My notes | 60+ minutes study

Aug. 17, 2021

I like to take some notes and relax, and learn a few things. 

The notes link is here

I looked up the website for Ladis 2009, and then the link is here to slides. 

Numbers everyone should know 

  • L1 cache reference 0.5 ns
  • Branch mispredict  5 ns
  • L2 cache reference 7 ns
  • Mutex lock/ unlock 25 ns
  • Main memory reference 100 ns
  • Compress 1K bytes with Zippy 3,000 ns
  • Send 2K bytes over 1 Gbps network  20,000 ns
  • Read 1 MB sequentially from memory 250,000 ns
  • Round trip within same datacenter 500,000 ns
  • Disk seek                                       10,000,000 ns
  • Read 1 MB sequentially from disk  20,000,000 ns
  • Send packet CA->Netherlands->CA 150,000,000 ns
Designing efficient systems 
Given a basic problem definition, how do you choose the "best" solution?
  • Best could be simplest, highest performance, easiest to extend, etc.
Important skill: ability to estimate performance of a system design
   - without actually having to build it!

Architectural view of the storage hierarchy

One server
DRAM: 16GB, 100ns, 20GB/s
Disk: 2TB, 10ms, 200MB/s

Rack Switch 
Local rack ( 80 servers)
DRAM: 1TB, 300us, 100MB/s
Disk: 160TB, 11ms, 100MB/s

Cluster (30+ racks)
DRAM: 30TB, 500us, 10MB/s
Disk: 4.80PB, 12ms, 10MB/s

 Back of the envelope calculations

How long to generate image results page (30 thumbnails)?

Design 1: Read serially, thumbnail 256K images on the fly
30 seeks * 10 ms/ seek + 30 * 256K /30 MB/s = 560 ms

Design 2: Issues reads in parallel:
10 ms/ seek + 256K read / 30 MB/s = 18 ms

(Ignores variance, so really more like 30-60 ms, probably)

Lots of variations:
  • caching (single images? whole sets of thumbnails?)
  • pre-computing thumbnails
  • ...
Back of the envelope helps identify most promising...

Know your basic building blocks
Core language libraries, basic data structure, protocol buffers, GFS, BigTable, indexing systems, MySQL, MapReduce, ...

Not just their interfaces, but understand their implementations (at least at a high level)

If you don't know what's going on, you can't do decent back-of-the-envelope calculations!

MapReduce
  • A simple programming model that applies to many large-scale computing problems
  • Hide messy details in MapReduce runtime library:
    • automatic parallelization
    • load balancing
    • network and disk transfer optimizations
    • handling of machine failures
    • robustness
    • improvements to core library benefit all users of library!
Typical problem solved by MapReduce
  • Read a lot of data
  • Map: extract something you care about from each record
  • Shuffle and Sort
  • Reduce: aggregate, summarize, filter, or transform
  • Write the results
Outline stays the same, map and reduce change to fit the problem

BigTable: Motivation
  • Lots of (semi-) structured data at Google
    • URLs:
      • contents, crawl metadata, links, anchors, pagerank, ...
    • Per-user data:
      • User preference settings, recent queries/search results, ...
    • Geographic locations:
      • Physical entities (shops, restaurants, etc.), roads, satellite image data, user annotations, ...
  • Scale is large
    • billions of URLs, many versions/page (~20K/ version)
    • Hundreds of millions of users, thousands of q/sec
    • 100TB+ of satellite image data
Basic data model
  • Distributed multi-dimensional sparse map (row, column, timestamp) -> cell contents
Rows are ordered lexicographically
Good match for most of our applications


BigTable status
  • Design/initial implementation started beginning of 2004
  • Production use or active development for 100+ projects:
    • Google Print
    • My Search History
    • Orkut
    • Crawling/indexing pipeline
    • Google Maps/Google Earth
    • Blogger
    • ...
  • Currently ~500 BigTable clusters
  • Largest cluster:
    • 70+ PB data; sustained: 10M ops/sec; 30+ GB/s I/O
Current work: Spanner
  • Storage & computation system that spans all our datacenters
    • single global namespace
      • Names are independent of location(s) of data
      • Similarities to Bigtable: tables, families, locality groups, coprocessors, ...
      • Differences: hierarchical directories instead of rows, fine-grained replication
      • Fine-grained ACLs, replication configuration at the per-directory level
    • support mix of strong and weak consistency across datacenters
      • strong consistency implemented with Paxos across tablet replicas
      • Full support for distributed transactions across directories/machines
    • much more automated operation
      • system automatically moves and adds replicas of data and computation based on constraints and usage patterns
      • automated allocation of resources across entire fleet of machines
Activities in world-wide systems
  • Challenge: automatic, dynamic world-wide placement of data & computation to minimize latency and/or cost, given constraints on:
    • bandwidth
    • packet loss
    • power
    • resource usage
    • failure modes
    • ...

  • User specify high-level desires:
    • "99%ile latency for accessing this data should be <50ms"
    • Store this data on at least 2 disks in EU, 2 in U.S. & 1 in Asia
Building applications on top of weakly consistent storage systems
  • Many applications need state replicated across a wide area 
    • For reliability and availability 
  • Two main choices:
    • consistent operations (e.g. use Paxos)
      • often imposes additional latency for common case
    • inconsistent operations
      • better performance/availability, but apps harder to write and reason about in this model
  • Many apps need to use a mix of both of these:
    • e.g. Gmail: marking a message as read is asynchronous, sending a message is a heavier-weight consistent operation
Building application on top of Weakly Consistent Storage Systems 
  • Challenge: General model of consistency choices, explained and codified
    • ideally would have one or more "knobs" controlling performance vs. consistency
    • "knob" would provide easy-to-understand tradeoffs
  • Challenges: Easy-to-use abstractions for resolving conflicting updates to multiple versions of a piece of state
    • Useful for reconciling client state with servers after disconnected operation
    • Also useful for reconciling replicated state in different data centers after repairing a network partition

  Further readings:
  1. Google File system, SOSP 2003
  2. Web search for a palnet: The Google Cluster Architecture, IEEE Micro, 2023
  3. OSDI 2004, MapReduce: Simplified Data processing on Large Clusters
  4. OSDI 2006, Bigtable: A distributed storage system for structured data 
  5. OSDI 2006, The Chubby Lock service for loosely-coupled distributed systems
  6. FAST 2007, Failure trends in a large disk drive population 
  7. EMNLP 2007, Large language models in Machine translation
  8. 2009, The datacenter as a computer: An introduction to the design of Warehouse-Scale machines
  9. 2009, PODC, Pregel: A system for large-scale graph processing 
  10. SEGMETRICS'09, DRAM Errors in the Wild: A Large-Scale Field study 
  11. Protocol buffers. http://code.goolge.com/p/protobuf/





    

Thursday, August 5, 2021

Bigtable: Harvard lecture notes | My 60+ minutes learning | Large distributed system

August 5, 2021

I like to take some time to learn Harvard lecture notes about Bigtable. Here is the link. 

The following is the lecture notes with my highlights. I also like to write down my study notes to help myself to be a better learner. 

Follow up on August 5, 2021 7:15 PM
I just could not believe that it is best lecture note I read in my whole life. Unbelievable!

Notes on Bigtable: A Distributed Storage System for Structured Data

The most influential systems publications of the 2000s may be the two first papers on Google’s internal cluster storage, GFS [1] and Bigtable [2]. GFS offers a file system-like interface, Bigtable a database-like interface; that is, GFS stores unstructured files (byte streams), and Bigtable stores structured data (rows, columns). But neither system uses a conventional interface. You read and write GFS files using a GFS API, and read and write Bigtable using a Bigtable API, not SQL.

Bigtable in particular is a delicious smorgasbord of data storage techniques, with a lot to teach us about building storage systems. On the other hand, several aspects of its design are sensitive to its deployment at Google, on top of GFS. To explain the design, we’ll pretend to build it up from first principles.

Reliable storage: durability and replication

Most any storage system aims to store data reliably, so that if a computer fails, the data can be recovered. We worry about both temporary failures, where a computer goes offline for a while but will come back, and permanent failures, where a computer dies. Network partitions, power blips, and program crashes generally cause temporary failures; hardware failure, fires, and sabotage generally cause permanent failures. We assume (with good reason) that temporary failures are more common and unpredictable than permanent ones.

To guard against power blips and program crashes, a system must store data on durable media, such as disks and flash memory. Only data stored on durable media will survive reboot. (Reboot is a magic solution for many temporary failures.)

But durable media cannot guard against permanent failures. That requires replication, where the system keeps multiple copies of the data: backups, basically. If the data is stored several times, on several geographically distributed computers, then only a major catastrophe will cause data loss.

Most (but, interestingly, not all) distributed systems use both durability and replication to store data reliably. For instance, each data modification might be written to at least three disks. If one disk fails, the data is proactively copied onto a new disk, so that at least three copies are usually available. That way, only three simultaneous permanent failures cause data loss. (Non-durable replication has not been considered sufficient since temporary failures—which are more common, and so might happen simultaneously—lose non-durable data.)

Most GFS files are replicated to three computers, which write them durably onto disks and flash.

Sequential storage

GFS was designed to store very large files that are generally accessed sequentially: starting from the first byte and proceeding in order. Sequential access is almost always the fastest way to access files on any storage system. Why?

  • Because hard disks are mechanical objects that spin. Reading data in a random order asks a disk’s mechanical “head” to jump around, a process called seeking. The head estimates the place to jump to and then must settle to get it right. A disk can do at most a couple hundred seeks a second. Sequential access (on sequentially-laid-out files) avoids seeking. Although in flash memory the seek penalty is much, much smaller, it still exists.
  • Because sequential access is predictable, all system caches have an easier job. The operating system can prefetch future data, dramatically speeding up future reads, simply by reading the next couple blocks of the file. The disk/flash itself can do the same thing with on-drive caches.

Structured storage

Bigtable, however, stores structured data, including large items (like web pages) and small items (like the text of a link). A typical Bigtable transaction might involve only a couple small data items, but many, many clients may access a Bigtable at a time. This offers both performance and correctness challenges. How can such a system scale?

Bigtable makes a couple data model choices relevant for our understanding.

  • Sparse hashtable. A Bigtable is essentially a sparse 3D hash table, where the dimensions are row names, column names, and versions (timestamps).
  • Strings. All Bigtable row names, column names, and data items are strings (sequences of characters). Bigtable has no true schema: everything’s a string.
  • Put, get, scan. Bigtable supports four fundamental operations: put (store a value in a row/column entry), get (return the value in a row/column entry), delete (delete a row/column entry), and scan (return many values from many row/column entries, in sorted order).

Building up Bigtable

We now describe roughly how Bigtable could have been designed, starting with the basics.

However, to make the issues clear, we’ll start a data model even simpler than Bigtable’s. Specifically, we’ll pretend that Bigtable started as a hash table, or key/value store, that maps string keys to string values. Here, a key combines the real Bigtable’s row and column names. Think of a key as the concatenation of those names (like “rowname|columnname”). We’ll see later why rows and columns are important to differentiate at the system level. But notice how far we can get without explicit columns: it may surprise you!

Basic reads and writes

  • Challenge: Efficiently yet reliably storing updates
    • Explanation: In disk storage efficient means sequential, so efficiently storing updates requires writing those updates sequentially. But updates arrive in random order, and must be stored as quickly as they arrive, since clients are waiting.
    • Solution: The only way to store updates sequentially is to order them sequentially: updates must be stored in a commit log, chronologically, as they arrive. This technique is ubiquitous in structured storage.
    • GFS note: GFS only provides reliable semantics for sequential log storage, which it supports via an “record append” operation.
  • Challenge: Efficiently supporting reads
    • Explanation: Logs are great for writing efficiently, but make no sense for reading (to read an item, a reader would have to scan the whole log to find the most recently written version). Most systems with log storage also maintain another data structure optimized for reads.
    • Solution: Bigtable servers store recent updates in memory, in a data structure called the memtable. Reads are quickly served out of the memtable. If a server crashes, its memtable can be reconstructed from the commit log.
  • Challenge: Data too big for memory
    • Explanation: Memtables work only as long as all data fits in memory, and servers rarely crash (restoring from log is slow). We need another durable data structure optimized for reads.

    • Solution: When a memtable gets too big or too old, Bigtable converts it into a durable structure called an SSTable. SSTables are optimized for reading. They store information about rows (keys) in lexicographic (dictionary) order, so scanning a table uses sequential access (fast). They are divided into 64KB segments, and a compact initial header specifies the initial key in each segment; thus a reader can seek to a key without reading the whole SSTable into memory.

      The memtable-to-SSTable process is called a minor compaction. It can happen in parallel with updates: Bigtable first starts a new memtable, then compacts the previous, frozen memtable in the background.

  • Challenge: Updates after minor compaction
    • Explanation: Converting to an SSTable does not stop the flow of updates. What should be done with them once the SSTable is on disk?

    • Conventional solution: Most databases solve this problem by maintaining a durable read/write data structure, usually a B-tree or variant (B+tree, B-link tree). Updates are first written to the log, which as in Bigtable is the primary commit point, and then lazily applied to the durable read/write structure. This works, but has some serious performance consequences. It is very hard to modify a read/write structure safely, and as the structure is modified, it inevitably drifts away from sequential layout.

    • Context: Bigtable’s storage layer, namely GFS, is ill suited for read/write structures. Not only was it designed for append—so in-place writes might be slow—but it doesn’t even provide consistency for in-place writes!

    • Solution: A particular Bigtable is implemented as an overlay stack of multiple tables. The memtable is on top, with the most recent updates; any value stored in the memtable has precedence over all other values. Underneath it are zero or more immutable SSTables: Bigtable never modifies an SSTable after it is created.

      To find a particular key, Bigtable checks these tables in reverse chronological order, using the first value it finds. Thus, each table acts as an overlay on top of older tables. To scan the database, Bigtable scans each of the tables in parallel, merge-sort-style; this is easy since each table is in sorted order. (If the same key appears in more than one table, Bigtable uses the value in the most recent table.) To update a key, Bigtable writes to the memtable. To delete a key, Bigtable must explicitly store a deletion record, or tombstone, that hides any lower occurrences of the key. (This resembles the use of tombstones in open-addressed hash tables.)

      (The overlay stack idea relates to log-structured merge trees [3] and read-optimized stores [4][5].)

  • Challenge: Garbage collection
    • Explanation: As updates and deletes collect, the stack of SSTables will get taller. This causes two problems. First, the lookup process takes O(t) time for a t-high stack. Second, past versions of updated data still take up space in the stack.

    • Solution: This is a garbage collection problem, and is solved garbage collection style. Periodically, Bigtable merges together several SSTables into one. In a merging compaction, the memtable and several recent SSTables are combined into a single SSTable. In a major compactionall SSTables are combined into a single SSTable.

      Why two types? A merging compaction might perturb updates a bit (since it consumes the memtable), whereas a major compaction need not (it does not appear to involve the memtable). A merging compaction will involve relatively less data (in SSTable stacks we expect the lower SSTables to contain more data), and is therefore faster to run. A merging compaction often needs to preserve tombstones to hide keys in lower, uncompacted SSTables, whereas a major compaction eliminates all deleted data.

      Note that many of these compaction operations can occur in the background as updates and lookups proceed in parallel. Whereas parallel databases often worry a lot about locking disciplines and scalability, Bigtable’s operations are naturally parallel: SSTables are immutable, and there is no need to obtain a lock before accessing an immutable object!

Scalability

  • Challenge: Data too large for a single computer
    • Context: The mechanisms we’ve described so far are really important to get right for any size database. But Bigtable is meant to scale to databases and client loads far too large for any single computer to handle, even if we assume that Bigtable’s underlying file system, GFS, scales perfectly.
    • Solution: Partition the Bigtable database among many servers. If there are n servers, give each server 1/n of the database. If we assume that each query touches just one key, then each server handles 1/n of the total query load. Linear scalability!
  • Challenge: Partitioning the key space
    • Context: How to split arbitrary string keys?

    • Solution: Divide key space into lexicographic ranges. If there are n servers, define pivot points x0 ≤ … ≤ xn, where x0 is the smallest possible string (the empty string) and xn is the largest possible string (∞). Then server i ∈ [0, n) handles all keys in the range [xixi+1).

      The portion of data stored on a particular server is called a tablet, and the server is called a tablet server. So a tablet consists of a commit log, a memtable, and zero or more SSTables. A tablet server can serve multiple tablets.

  • Challenge: Locating servers
    • Context: How can clients find the server responsible for a key?
    • Solution: Store this information in Bigtable itself! A set of METADATA tablets, arranged B-tree-style, list the locations of all other tablet servers in the system. These are indexed by table name and key range. The location of the topmost METADATA tablet is stored in Chubby, a reliable component outside of GFS. So a client can find a tablet by contacting Chubby and walking through METADATA tablets. In practice, the client caches METADATA tablet data.
  • Challenge: Compensating for failures
    • Context: What happens if a tablet server dies? The data is safe, since it’s stored reliably in GFS according to the procedures above, but we need a Bigtable server to coordinate the pieces and actually serve data.
    • Solution: A distinguished master component monitors failed tablet servers and reassigns their tablets as necessary. Tablet servers register themselves as available for tablets; the master then assigns tablets to servers, recording its choices in the METADATA tablets so clients can find them. This master is a centralized component, but note that it’s not on the critical path for client requests—clients can read METADATA tablets independent of the master. However, a working master is required to assign tablets to servers. A separate system component, the “cluster management system,” restarts the master as necessary.
  • Challenge: Updating the partitioning
    • Context: Data isn’t uniformly distributed. A stream of updates and deletes may make some tablets (partitions) grow too large to be effective, or get so small that they’re not worthwhile. Partition points should be updated based on the characteristics of the data.
    • Solution: Tablet servers split themselves, entering new split points in the METADATA tablets. All other changes (tablet merges, new tables, schema changes) are managed by the master.

Transaction support

  • Challenge: Applications want atomic, consistent updates and transactions
    • Context: Everyone loves ACID (Atomic, Consistent, Isolated, Durable) transactions. Can Bigtable provide this notion of consistency? And if not, can it provide any notion of consistency?

    • Solution: Bigtable already provides durability and atomicity through the commit log. But conventional databases provide arbitrary-sized transactions: a single transaction can modify the entire database in one atomic step. In Bigtable’s distributed context, however, arbitrary transactions would require coordination among many tablet servers. This coordination—basically, locking—would compromise scalability. So Bigtable punts. Bigtable does not support arbitrary transactions.

      Coordination is easy within a single tablet server, though. So Bigtable does support limited “read-modify-write”/“compare-and-swap” type transactions that touch one tablet server at a time. However, Bigtable would need to worry about splits and other concurrent updates during a transaction. So Bigtable ensures that the data relevant to one transaction will never be split across two tablets. An easy way to do that would be to limit transactions to support exactly one key: basically, compare-and-swap on a single key/value pair.

  • Challenge: Single-key transactions are too limiting
    • Context: Single-key transactions are easy to implement, but are too limiting for most clients. There’s a tremendous flexibility difference between “atomic integers” and “atomic structure operations” on arbitrary structures.

    • Solution: Split the key into two parts, a row and a column. Partition data based on row key, so that each tablet handles a contiguous range of rows. Then we know that any split will keep all of a row’s data together, and Bigtable can support transactions that operate on a single row at a time without too much work. This is much better than single-key transactions, because a row can have arbitrarily many columns.

      (Of course, it’s very likely that Bigtable had rows and columns from the start, but they weren’t exactly necessary until now.)

      Unlike conventional databases, Bigtable rows can have arbitrarily many columns, and different rows can have different sets of columns. This shows how close the Bigtable column idea is to general string keys. It also makes some cool programming tricks possible; for instance, consider Figure 1 [2]. A single row has “anchor:cnnsi.com” and “anchor:my.look.ca” columns, which a conventional database would store in a separate table (a “page_anchor” table with a two-column unique key, “page_url + “anchor_url”). Bigtable general columns are especially useful since Bigtable transactions are limited. Many updates that, in a conventional database, would require cross-table coordination, Bigtable can handle with atomic row updates.

  • Challenge: Support transactions in the future

    • Context: Many applications don’t need full transaction support, so maybe it’s fine that Bigtable doesn’t provide it. But it would rock to build some sort of system on top of Bigtable that supported transactions. Is there anything that Bigtable might need to support now that would simplify transaction implementation later?

    • Solution: Multiple versions of data. Bigtable is willing to store many versions of a value for a given row/column pair. These versions are indexed by timestamp. Given both timestamps and read-modify-write operations, Bigtable can support both locks and multi-version concurrency control like snapshot isolation.

      Of course, timestamps are useful for other, related purposes. Clients can roll their own “transactions” by examining only information from a specific timestamp range, or check recent changes by scanning for updates after a given timestamp. Bigtable supports configurable garbage collection of old data: compactions can throw out old versions.

Optimizations

We only describe a limited set; see the paper for more.

  • Challenge: Handle more data per server

    • Context: Reading from disk is slow and proportional to the amount of data read. Memory is also a bottleneck.

    • Solution: Compress SSTables, using new, speed-optimized compression algorithms. Compression is done block by block, so a tablet server can scan to a given SSTable block without uncompressing all prior blocks.

  • Challenge: Skip irrelevant data

    • Context: The Bigtable representation encourages application designers to combine different classes of information into single rows, where conventional databases would split those information classes into different tables. (Consider Figure 1, which stores both multiple versions of a crawled web page and many small text snippets.) As a result, rows are very large and can combine big data items with small ones. An application might be interested in scanning over just the small data items (e.g., the text snippets), but the SSTable representation described so far would require tablet servers to read all data items into memory for each row.

    • Solution: Divide columns into subsets, and store column subsets separately. Specifically, split each column into two parts, the column family and the column qualifier. Multiple column families can be grouped together into locality groups. All columns in the same locality group are stored in the same SSTable stack, but columns in different families are stored in different SSTable stacks. Then, we can group the Figure 1’s small text snippets together, and scan them all without reading web page data into memory.

      This design means that a single tablet can comprise multiple SSTable stacks. (It appears that the memtable is shared among locality groups, and a minor compaction can split a single memtable into several SSTables, at most one per stack.) But this doesn’t affect atomic update consistency. Since all the SSTables for a given row are always stored by the same tablet server, the tablet server can still easily update rows atomically.

  • Challenge: A t-high SSTable stack takes O(t) SSTable reads to test for a key

    • Solution: Bigtable can associate a Bloom filter with each SSTable. The Bloom filter is a conservative set representation that takes very little memory. It can confirm that a particular key is definitely not in an SSTable, but it can’t say for sure whether a key is actually present. That is, it can give false positives but never false negatives, which is why it’s conservative. If a tablet server keeps all SSTables’ Bloom filters into memory, it can often avoid failed key lookups in SSTables. This can reduce the number of SSTable reads to below O(t), although obviously the number of Bloom filter reads is still O(t).

Bigtable as a whole

Here’s an overview of the whole Bigtable system as we’ve described it.

Cluster level

  • Bigtable cluster has exactly one Chubby instance.
  • Bigtable cluster has at most one Bigtable master. (If the master fails, it is restarted by other system components.)
  • Bigtable cluster has one tablet hierarchy.
  • The tablet hierarchy is rooted at a single root tablet.
  • tablet hierarchy has many METADATA tablets, which store location information for other user tablets.
  • Each user tablet belongs to exactly one Bigtable.
  • All of a Bigtable’s tablets are part of the same cluster and in the same hierarchy.

Tablet level

  • Each tablet is served by at most one tablet server.
  • tablet server can serve many tablets. (The master assigns tablets to servers.)
  • Each tablet comprises one or more SSTable stacks, one per locality group, as well as a memtable and a portion of a commit log.
  • Commit logs and tablet servers correspond one-to-one (except during failures).
  • Each commit log contains data from one or more tablets (see “Commit log implementation” in §6).
  • Each memtable contains data from exactly one tablet.
  • Each SSTable stack comprises zero or more SSTables.
  • Each SSTable comprises one or more 64KB blocks, plus an optional Bloom filter. The blocks can be compressed; each block is compressed separately.
  • Minor compactions change a single memtable into a set of new SSTables, at most one per stack. Minor compactions increase SSTable stack height.
  • Merging compactions are minor compactions that also combine some of the upper SSTables on the previous stacks with the new SSTables. Merging compactions can reduce SSTable stack height.
  • Major compactions combine all existing SSTables together. Major compactions cut SSTable stack height down to one.

Row level

  • Each tablet contains data for a contiguous range of rows.
  • Each column family belongs to one locality group.
  • Each locality group contains one or more column families.
  • Each column belongs to a single column family.
  • transaction accesses data in at most one row.
  • Each row can contain data in many columns. Different rows can contain different columns.

Client level

  • client accesses many tablet servers. A client will practically never contact the master, or Chubby.
  • client can create arbitrary rows and arbitrary columns in pre-existing column families. Only a Bigtable’s administrator can change its column families.
  • Access control decisions are made at the column family level.

Comparison with conventional databases

  • Database ~ Bigtable
  • Table ~ Column family
  • Primary key ~ Row
    • In Bigtable, all “tables” (column families) always have the same primary key.
  • B-tree node ~ Tablet
    • In a conventional database a B-tree node stores a row range from a single table, whereas a tablet contains row ranges for many column families.
  • Transaction ~ Atomic row update
  • Schema ~ Column family schema

Contributions

Bigtable not only introduced an interesting data model (rows, columns, column families, timestamps, atomic row updates), it also combined a large number of interesting and useful data representation techniques (mutable stacks of immutable SSTables, Bloom filters, compressed tablets), some of them new. The paper offers a deep set of systems techniques and obviously good engineering. The Chubby/master/tablet-server interactions (which we didn’t particularly focus on above) show that single-master systems can avoid bottlenecks and scale tremendously.


  1. “The Google file system,” Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung, in Proc. 19th SOSP, 2003 (ACM Digital LibraryGoogle Research Publications)

  2. “Bigtable: A distributed storage system for structured data,” Fay Chang, Jeffrey Dean, Sanjay Ghemawat, Wilson C. Hsieh, Deborah A. Wallach, Mike Burrows, Tushar Chandra, Andrew Fikes, and Robert E. Gruber, in Proc. 7th OSDI, Nov. 2006 (Via USENIXACM Digital LibraryGoogle Research Publications)

  3. “The log-structured merge-tree (LSM-tree),” Patrick O’Neil, Edward Cheng, Dieter Gawlick, and Elizabeth O’Neil, Acta Informatica 33(4):351–385, 1996.

  4. “Performance tradeoffs in read-optimized databases,” Stavros Harizopoulos, Velen Liang, Daniel J. Abadi, and Samuel Madden, in Proc. VLDB ’06, pages 487–498, 2006.

  5. “Rose: Compressed, log-structured replication,” Russell Sears, Mark Callaghan, and Eric Brewer, in Proc. VLDB ’08, August 2008.

 Actionable items:

  1. Continue to study memTable and SSTables. 
  2. Continue to work on more readings related to lecture notes related to BigTable. 
  3. Go back to read the original paper related to BigTable if I have time. 

Follow up

Jan. 6, 2022
I like to work on bigtable lecture notes after I spent time to read article related to bigTable field promotion technique.