Sheet 1
NoSQL Ecosystem
  Definition
    Not a tool
    Ecosystem composed of several tools
    Not only SQL
    Suitable for specific problem
      Large amount of data
      Massive workload
      Difficult data modeling decision
  SQL Brief
    Declarative language
    Components
      Query optimizer
    Rational database
      Relational data model
      Data table joining
      Data aggregation
    Drawback
      Unpredictable workload or cost
      Not adaptive to less-structured data
      Growing data leads to table partition
    Guarantees
      Transactional semantics
        All-or-nothing
      Consistency
        No error for the reads after a write
      Durability
        The data are in a stable storage (not cache)
  Inspiration
    Google's BigTable
    Amazon's Dynamo
    HBase
    Voldemort
    Cassandra
    Use them to illustrate the approaches adopted by the NoSQL ecosystem
  Other Roadmap
    Data and query model
    Durability
    Scalability
    Partitioning
    Consistency
    Transactional semantics
    Single-server performance
    Analytical workloads
    Not fully covered in this article
  Models (overview)
    Data model
      How the data is logically organized
      Types
        Relational model
        Key-oriented storage model
        Various graph models
    Quuery model
      How the data can be retrieved and updated
      Types
        SQL
        Key lookups
        MapReduce
  Consistency
    Achieve consistency while replicating
      Replica will crash
      Network partition with two replica sets
      Message delay or lost
    CAP
      Consistency
      Availability
      Partition tolerance
        All NoSQL sys should have this
      We can only achieve two of them
      Choose one from these two
    Strong consistency
      All replicas will agree with the value being read
        N machines
        Need W to confirm writes before returning to user
        Need R to confirm data reading before returning to user
        R+W>N if strong consistency
          Usually R+W=N+1
          Or W=N, R=1
      If not (timeout or error), fix it or return error to user
        System not available for a while
    Eventual consistency
      Allow R+W <= N
      Versioning and conflicts
        Vector clock
      Conflict resolution
        Implemented by the developers
          Merged the keys
            Shopping cart application
          Resolve the conflicts manually
            Collaborative document edition
        Cassandra: most recent timestamped keys
        Riak: allow both methods
        CouchDB: hybrid, notify conflict, return a latest copy before reparation
      Read repair
        Resolve if the coordinator see different version of data from the replicas
      Hinted handoff
        If a node goes down, the write load goes to a backup server
        When the node recover, the backup server forward the writes to it
      Anti-entropy
        Node is down, backup server is also down, the data should also be safe
        Replicas exchanges Merkle Trees
          Identify out of sync key ranges
      Gossip
        Exchange message (gossip) to get the health report of the nodes in the system
  Scalability
    Vertical scalability (scale up)
    Horizontal scalability (scale out)
      Linear scalability
    Sharding
      Scale without sharding
        Read from replicas
          Less demands for content freshness
        Caching
          Handling large number of read-only requests
      Sharding through coordinators
      Consistent hash rings
        Consistent hashing
        Distributed hash tables (DHTs)
        Replicating data
          Multiple replicas for each key range
        Achieving better distribution
          "Virtual nodes" to balance the key ranges
          Adjust the location of servers based on historic load
      Range partitioning
        Two closed keys are likely to appear in the same partition
          Monitor the traffic, control the range of servers dynamically
        The BigTable way
          Hierarchical metadata
        Handling failures
          Master server
            Could be single point of failure
but can go down temporarily
          BigTable: Chubby, distributed locking system (Consistency)
          Open source version: ZooKeeper
        Range partitioning-based NoSQL projects
          HBase = BigTable hierarchical approach+HDFS
          MongoDB config nodes=hybrid(BigTable, config management)
          Cassandra: map key to the server naturally
          Twitter's Gizzard: Hierarchies of any depth
            Replication: send the updates to multiple machines
        Which partitioning scheme to use?
Hash- or range-based
          Frequently performing range scans over the keys
            Use range-based
            Less network overhead
          Infrequent range scans, want reasonable data distribution
            Use hash-based
  Data Durability
    Single-server durability
      Limiting random writes
      Increasing sequential writes
      Techniques to improve performance
        Control `fsync` frequency
        Increase sequential writes by logging
        Increase throughput by grouping writes
    Multi-server durability
      Master-slave approach to replicating data (Redis)
      Replica sets (MongoDB, HBase)
      N replica machiens, W (W<N) confirmation before returning True/False
      Usually configurable by the user
  Data and query model
    Key-based NoSQL data model
      Key-value stores
      Key-data structure stores
      Key-document stores
      BigTable Column Family stores
    Graph storage
      Differ from other stores
        Data models
        Data traversal
        Querying patterns
    Complex queries
      Index data based on any number of properties
      Scanners to iterate over a column family
      Create different views of the data
      Run MapReduce tasks to support complex lookups and updates
      Bindings to Hadoop/Spark/MapReduce
    Transactions
      ACID
        Atomic
          All-or-nothing happens
        Consistency
          Uncorrupted
        Isolation
          No dead lock/live lock
        Durability
          Committed transaction is stored safely
        NoSQL has perfomance over full guarantee but not at key level.
    Schema-free stroage
      Sloppy-schema NoSQL
  Interchangeable terms
    Partition and sharding
    Machine, server, and node
    Cluster and ring
  Facts
  External links
SQL's expressiveness makes it challenging to reason about the cost of each query, and thus the cost of a workload. While simpler query languages might complicate application logic, they make it easier to provision data storage systems, which only respond to simple requests.
The relational data model is strict: the schema assigned to each table specifies the data in each row. If we are storing less structured data, or rows with more variance in the columns they store, the relational model may be needlessly restrictive. Similarly, application developers might not find the relational model perfect for modeling every kind of data. For example, a lot of application logic is written in object-oriented languages and includes high-level concepts such as lists, queues, and sets, and some programmers would like their persistence layer to model this.
To avoid JOINs having to cross the network in order to get data in different tables, we will have to denormalize it. Denormalization stores all of the data from different tables that one might want to look up at once in a single place. This makes our database look like a key-lookup storage system, leaving us wondering what other data models might better suit the data.
One row decreases, the other one increases, typically in bank transactions
do all replicas of a piece of data always logically agree on the same version of that data by the time you read it? (This concept of consistency is different than the C in ACID.)
Opting for consistency means that your replicated data will not be out of sync across replicas. An easy way to achieve consistency is to require that all replicas acknowledge updates. If a replica goes down and you can not confirm data updates on it, then you degrade availability on its keys. This means that until all replicas recover and respond, the user can not receive successful acknowledgment of their update operation. Thus, opting for consistency is opting for a lack of round-the-clock availability for each data item.
Do replicas respond to read and write requests regardless of how many replicas are inaccessible?
Opting for availability means that when a user issues an operation, replicas should act on the data they have, regardless of the state of other replicas. This may lead to diverging consistency of data across replicas, since they weren't required to acknowledge all updates, and some replicas may have not noted all updates.
Can the system continue to operate even if some replicas temporarily lose the ability to communicate with each other over the network?
All replicas remain in sync
The replicas are allowed to get out of sync but eventually catch up with one-another.
For example, if servers A, B, and C are the three replicas of some key, the vector clock will have three entries, (N_A, N_B, N_C), initialized to (0,0,0).
Each time a replica modifies a key, it increments its counter in the vector. If B modifies a key that previously had version (39, 1, 5), it will change the vector clock to (39, 2, 5).
When another replica, say C, receives an update from B about the key's data, it will compare the vector clock from B to its own. As long as its own vector clock counters are all less than the ones delivered from B, then it has a stale version and can overwrite its own copy with B's.
If B and C have clocks in which some counters are greater than others in both clocks, say (39, 2, 5) and (39, 1, 6), then the servers recognize that they received different, potentially unreconcilable updates over time, and identify a conflict.
The coordinator may still notice that some of the replicas are out of sync. The Dynamo paper suggests, and Cassandra, Riak, and Voldemort implement, a technique called read repair for handling such situations. When a coordinator identifies a conflict on read, even if a consistent value has been returned to the user, the coordinator starts conflict-resolution protocols between conflicted replicas. This proactively fixes conflicts with little additional work. Replicas have already sent their version of the data to the coordinator, and faster conflict resolution will result in less divergence in the system.
If one of the replicas for a key does not respond to a write request, another node is selected to temporarily take over its write workload. Writes for the unavailable node are kept separately, and when the backup node notices the previously unavailable node become available, it forwards all of the writes to the newly available replica. The Dynamo paper utilizes a 'sloppy quorum' approach and allows the writes accomplished through hinted handoff to count toward the W required write acknowledgments. Cassandra and Voldemort will not count a hinted handoff against W, and will fail a write which does not have W confirmations from the originally assigned replicas. Hinted handoff is still useful in these systems, as it speeds up recovery when an unavailable node returns.
A Merkle tree is a hierarchical hash verification: if the hash over the entire keyspace is not the same between two replicas, they will exchange hashes of smaller and smaller portions of the replicated keyspace until the out-of-sync keys are identified. This approach reduces unnecessary data transfer between replicas which contain mostly similar data.
as distributed systems grow, it is hard to keep track of how each node in the system is doing. The three Dynamo-based systems employ an age-old high school technique known as gossip to keep track of other nodes.
Periodically (every second or so), a node will pick a random node it once communicated with to exchange knowledge of the health of the other nodes in the system. In providing this exchange, nodes learn which other nodes are down, and know where to route clients in search of a key.
Doubling the number of machines will double the query capacity of the system.
The act of splitting your read and write workload across machines to scale out the storage system.
Sharding is fundamental to the design of many systems, namely Cassandra, HBase, Voldemort, and Riak, and more recently MongoDB and Redis.
Sharding means that no one machine has to handle the write workload on the entire dataset, but no one machine can answer queries about the entire dataset. Most NoSQL systems are key-oriented in both their data and query models, and few queries touch the entire dataset anyway.
Sharding is typically key-based
Sharding adds system complexity.
Do not shard until you have to.
Many storage systems see more read requests than write requests. A simple solution in these cases is to make copies of the data on multiple machines. All write requests still go to a master node. Read requests go to machines which replicate the data, and are often slightly stale with respect to the data on the write master.
If you are already replicating your data for multi-server durability in a master-slave configuration, as is common in Redis, CouchDB, or MongoDB, the read slaves can shed some load from the write master. Some queries, such as aggregate summaries of your dataset, which might be expensive and often do not require up-to-the-second freshness, can be executed against the slave replicas. Generally, the less stringent your demands for freshness of content, the more you can lean on read slaves to improve read-only query performance.
Because Memcached is designed for caching, it does not have as much architectural complexity as the persistent solutions for scaling workloads. Before considering more complicated solutions, think about whether caching can solve your scalability woes.
Caching is not solely a temporary band-aid: Facebook has Memcached installations in the range of tens of terabytes of memory!
CouchDB + Lounge/BigCouch
Twitter: Gizzard: Tree-based by keys.
NoSQL systems built around the principles of Amazon's Dynamo adopted this distribution technique, and it appears in Cassandra, Voldemort, and Riak.
Using modulo operation.
Each virtual node hashes to a different value, giving it more opportunity to manage keys distributed to different parts of the keyspace. Voldemort takes a similar approach, in which the number of partitions is manually configured and usually larger than the number of servers, resulting in each server receiving a number of smaller partitions.
The metadata size will be largely reduced. Instead of storing each keys: server pair, the metadata can only contain range [start, end].
The master adjusts which tablet server serves which tablets at any time.
Because this metadata can get large, the metadata table is also sharded into tablets that map key ranges to tablets and tablet servers responsible for those ranges. This results in a three-layer hierarchy traversal for clients to find a key on its hosting tablet server
The BigTable paper explains that this 3-level hierarchy can accommodate 261 bytes worth of storage using 128MB tablets.
The master is a single point of failure in the BigTable design, but can go down temporarily without affecting requests to tablet servers. If a tablet server fails while serving tablet requests, it is up to the master to recognize this and re-assign its tablets while requests temporarily fail.
These configuration nodes stay in sync through a protocol called two-phase commit, and serve as a hybrid of BigTable's master for specifying ranges and Chubby for highly available configuration management. Separate routing processes, which are stateless, keep track of the most recent routing configuration and route key requests to the appropriate storage nodes. Storage nodes are arranged in replica sets to handle replication.
Range partitioning requires the upfront cost of maintaining routing and configuration nodes, which can see heavy load and become central points of failure in the absence of relatively complex fault tolerance schemes.
Done well, however, range-partitioned data can be load-balanced in small chunks which can be reassigned in high-load situations. If a server goes down, its assigned ranges can be distributed to many servers, rather than loading the server's immediate neighbors during downtime.
Different NoSQL systems trade off durability guarantees for performance.
Memcached: No durability but very fast;
Redis: fsync after every update or fsync every N seconds, determined by the developer.
NoSQL use B+Trees to quickly retrieve data from disk, but may cause random writes. To reduce this, Cassandra, HBase, Redis, and Riak append updates to a sequentially-written file called a log.
(Similar to message queues?)
While NoSQL systems such as MongoDB perform writes in-place in their data structures, others take logging even further.
- Cassandra and HBase use a technique borrowed from BigTable of combining their logs and lookup data structures into one log-structured merge tree.
- Riak provides similar functionality with a log-structured hash table.
- CouchDB has modified the traditional B+Tree so that all changes to the data structure are appended to the structure on physical storage.
These techniques result in improved write throughput, but require a periodic log compaction to keep the log from growing unbounded.
Cassandra groups multiple concurrent updates within a short window into a single fsync call. This design, called group commit, results in higher latency per update, as users have to wait on several concurrent updates to have their own update be acknowledged. The latency bump comes at an increase in throughput, as multiple log appends can happen with a single fsync. As of this writing, every HBase update is persisted to the underlying storage provided by the Hadoop Distributed File System (HDFS)15, which has recently seen patches to allow support of appends that respect fsync and group commit.
The key lookup model is beneficial because it means that the database has a consistent query pattern—the entire workload consists of key lookups whose performance is relatively uniform and predictable. Profiling to find the slow parts of an application is simpler, since all complex operations reside in the application code. On the flip side, the data model logic and business logic are now more closely intertwined, which muddles abstraction.
Stored in a blob containing JSON or a binary format. Examples: Protocol Buffers, Thrift, Avro.
Most popular example: Redis.
Supported data structures: integer, string, list, set, and sorted set.
Operation: set/get/delete/increment/decrement/push/pop
Examples: CouchDB, MongoDB, Riak.
Usually the data is stored in JSON or JSON-like formats.
Supports lists and dictionaries, can have a nested structure.
Pro: Application developers have a lot of freedom in modeling their documents, but application-based query logic can become exceedingly complex.
HBase and Cassandra base their data model on Google's BigTable.
In this model, a key identifies a row, which contains data stored in one or more Column Families (CFs). Within a CF, each row can contain multiple columns. The values within each column are timestamped, so that several versions of a row-column mapping can live within a CF.
Column Families = hash(row ID, CF, column, timestamp)?
Pros:
1. modeling historical data with timestamps
2. Do not need an explicit NULL for those null columns.
Cons:
columns which have few or no NULL values must still store the column identifier with each row, which leads to greater space consumption.
Cassandra:
Cassandra introduces the notion of a supercolumn within each CF to allow for another level of mapping, modeling, and indexing. It also does away with a notion of locality groups, which can physically store multiple column families together for performance reasons.
Example: HyperGraphgDB, Neo4J
This push for correctness results in often-unexpected performance characteristics, where a slow transaction might cause an otherwise quick one to wait in line.
- Voldemort, Cassandra, Riak are based on Dynamo (?)
http://hbase.apache.org/
http://project-voldemort.com/
http://cassandra.apache.org/
http://code.google.com/p/protobuf/
http://thrift.apache.org/
http://avro.apache.org/
http://www.oracle.com/technetwork/database/berkeleydb/overview/index.html
http://redis.io/
http://couchdb.apache.org/
http://www.mongodb.org/
http://www.basho.com/products_riak_overview.php
http://www.hypergraphdb.org/index
http://neo4j.org/
http://memcached.org/
http://hadoop.apache.org/hdfs/
http://github.com/twitter/gizzard
http://hadoop.apache.org/zookeeper/