SD General Flashcards
What are the common use cases that event order are really important
Payments transaction.
Video conferencing.
Vehicle control systems.
Online gaming.
Sensor metrics publishing and changes capture.
Why we don’t want to store database in the application machines?
Bottleneck; independent scaling.
Typically application servers become a bottleneck to the system before the actual accessing to the data does.
What is write ahead log and why do we need it
Recovery from crash - data durability. RAM lose data after crash.
This is an append-only file to which every DB modification must be written before it can be committed to the DB. When the database comes back up after a crash, this log is used to restore the DB back to a consistent state.
Write ahead log is completely on disk so it’s durable, we can just replay the records in the WAL to repopulate the data.
Atomicity - revert some writes per errors.
Durability and consistency - replay the logs and rebuild the state.
———————
In database systems that use Write-Ahead Logging (WAL) and implement row-level locking, the behavior regarding when writes are recorded in the WAL relative to locking can vary depending on the specific database’s implementation and configuration. However, there are common patterns and practices that can be generalized to understand how most systems likely behave.
-
Acquiring the Lock:
- Before any write operation (update, delete, or insert) can be applied to a row in a database, the transaction that intends to make the change must first acquire the appropriate lock on that row. This lock ensures that no other concurrent transactions can modify the same row until the lock is released.
-
Writing to the WAL:
- Once the lock is successfully acquired and the row is effectively under the control of the current transaction, the changes are first recorded in the WAL. This step is crucial for ensuring data durability and atomicity; the database logs the intention to make changes before any actual modifications are applied to the data files.
- The WAL entry typically includes information necessary to redo the operation (in case of a crash after the WAL record is written but before the changes are committed) and to undo the operation (in case the transaction needs to be rolled back).
-
Applying Changes to the Data:
- After the WAL is updated, and the changes are securely logged, the transaction then applies these changes to the actual data stored in the database. This step is performed while the row lock is still held, preventing other transactions from accessing the row until the current transaction is complete.
-
Committing the Transaction:
- The transaction can be committed only after the WAL records are safely written to disk. Once the commit is successful, the locks can be released, allowing other transactions to access or modify the row.
- Order of Operations: The crucial aspect is that the WAL is written to after acquiring the necessary locks but before the changes are committed and the locks are released. This ensures that the database can always recover to a consistent state, as the WAL contains all the information required to redo or undo transactions up to the last known good state.
- Handling Deadlocks and Delays: If a lock cannot be acquired because another transaction holds a conflicting lock, the transaction will typically wait until the lock is available. During this waiting period, nothing is yet written to the WAL. If a deadlock is detected (where two or more transactions are waiting on each other indefinitely), one of the transactions will be chosen as a victim and rolled back to resolve the deadlock.
- Performance Implications: The need to acquire locks before writing to the WAL can impact performance, particularly in high-contention environments where many transactions are competing for the same rows. Database systems often implement sophisticated concurrency control mechanisms to minimize locking delays and optimize transaction throughput.
In summary, in systems using WAL and row-level locking, the write to the WAL occurs after the lock on the row is successfully acquired but before the transaction is committed. This sequence ensures that all changes logged in the WAL are guaranteed to be applied atomically and durably, maintaining the integrity and consistency of the database even in the face of system failures or crashes.
What is read and write time complexity of hash indexes? Why hash indexes are only stored in memory? Why range query is expensive for hash index?
Redis for key-value retrieval.
Riak for key-document retrieval.
PostgreSQL, MySQL for exact match queries.
Cassandra use it to map data to nodes using hash value of partition keys.
Basically hash map.
O(1) write and read for hash function, because memory or RAM is O(1) for random access by direct addressing.
Optimized for point queries, where we need to find rows with specific keys.
Subject to the requirement that all indexes can fit into memory. RAM is expensive, less, not durable.
Hash key is randomly and evenly distributed, and inherently lacking of order. Need to scan the entire hash table to find all keys that falls into the range.
Hash indexes are not good for hard disk due to random distribution of data blocks on disk, and therefore lack spatial locality, related data are not stored closer to each other. The disk arm or pointer needs to jump around.
Rehashing and redistribution leads to disk I/O overhead.
What is branching factor in B tree and how does it affect the time complexity of b-tree
MySQL, PostgreSQL, Oracle, MongoDB.
Number of children a node can have. Typically it is several hundred, depends on the storage to store page pointers and the range boundaries. Most databases can fit into B-tree of 3 or 4 levels. A 4 level tree of 4 KB pages with branching factor 500 can store up to 256 TB.
The higher the branching factor is, the less height the tree has, and therefore improving the read and write efficiency.
However, a large node can increase memory overhead, because some keys that act as separators can be stored in memory to speed up get guide to the correct leaf node.
DDIA 81
Why do we still need write ahead logs for B-tree indexes?
DDIA 82.
Operations requiring several pages to be overridden.
If the node crashes when you are partially done with page splits, you are ended up with a corrupted indexes.
How do we delete a record from the log structured file segments in disk, SST or simple unsorted segments
Append a special deletion record sometimes called a tombstone.
LSM tree + SST
Advantages over hash index + log structure files
Paper that proposed this idea came out at 1996.
Apache Cassandra - minimize disk I/O and maximize write throughput.
RocksDB, LevelDB, Google Bigtable
LSM tree is also in memory, same as hash indexes. It can be either red black trees or AVL trees. They are also called memtable.
Inorder traversal and write to disk as sorted string table when it gets full ( a few MBs). Every-time when we write memtable to SSTable, we can discard the corresponding WAL.
In Lucene, indexing engine, the mapping from term to postings list is kept in SSTable like sorted files, which are merged in the background as needed.
DDIA 75
LSM tree and SSTable optimization
Sparse index - the offsets of keys in the log segment.
Bloom filters - what keys does not appear in the database. Never provide false negative - if it says a key does not exist, it would definitely not exist.
Why is LSM tree + SST is a middle ground between hash index and b-tree
Not bounded by memory like hash indexes; can do range query that hash indexes are bad at
Better write performance than b tree while read performance not as good as b tree.
It requires extra CPU bandwidth for compaction
What is address index? Why do we need it sometimes?
Index maps to disk addresses. Like the sparse index we discussed for SSTable.
It’s a type of non clustered index. Don’t store entire row or duplicate data when we have indexes on different column to minimize the data we store.
What is multi-dimensional index and why do we need it?
Like a combo index; composite index.
Example: geospatial dataset, latitude and longitude.
For example, PostGIS implements geospatial indexes as R-trees using PostgreSQL’s Generalized Search Tree indexing facility.
An interesting idea is that multi-dimensional indexes are not just for geographic locations. For example, on an ecommerce website you could use a three-dimensional index on the dimensions (red, green, blue) to search for products in a certain range of colors, or in a database of weather observations you could have a two-dimensional index on (date, temperature) in order to efficiently search for all the observations during the year 2013 where the temperature was between 25 and 30℃. With a onedimensional index, you would have to either scan over all the records from 2013 (regardless of temperature) and then filter them by temperature, or vice versa. A 2D index could narrow down by timestamp and temperature simultaneously. This technique is used by HyperDex.
DDIA 87
What is atomicity
All or nothing principle. All writes for a transaction should succeed or none of them succeed
Example - exchange of money
DDIA 223
What is database consistency
If the transaction fails it fails gracefully, no invariants should be broken - foreign key constraint, unique constraint and check constraint.
Check constraint - specific rules beyond basic data types constraint; businesses rules within the database schema.
For example, employee hiring date always greater than date of birth.
In accounting, credits and debits across all accounts should be balanced.
Atomicity focus on transactions are executed as indivisible units;
consistency focus mainly on maintaining the integrity of database in line with its invariants, rules and constraints.
Application replies on database atomicity and isolation properties in order to achieve consistency, but it’s not up to the database alone.
DDIA 225
What is isolation
No race conditions.
Concurrently executing transactions are isolated from each other.
DDIA 225
What is Durability
No data lost.
Once a transaction has committed successfully, any data it has written will not be forgotten, even if there is a hardware fault or database crashes.
DDIA 226
What is dirty writes? Give an example
Overwrite uncommitted data.
Two transactions concurrently try to update the same object in a database.
If the earlier write is a part of a transaction that has not yet committed and the later write overwrites an uncommitted value.
DDIA 235
What is dirty read?
A transaction has not yet committed or aborted and another transaction see the uncommitted data.
DDIA 234
—————————
Dirty reads are problematic because they can lead to inconsistencies, unreliability, and potential data corruption in a database. When a transaction reads data that has been modified but not yet committed by another transaction, it risks basing its operations on temporary, potentially invalid data. This can result in several issues:
-
Inconsistencies:
- If the transaction that made the uncommitted change is rolled back, the dirty read transaction would have read and possibly acted on data that never becomes permanent.
-
Cascading Rollbacks:
- If a dirty read occurs and the initial transaction is rolled back, subsequent transactions that read the uncommitted data might also need to be rolled back, causing a cascade of rollbacks.
-
Unreliability:
- Applications relying on database transactions may become unreliable if they frequently operate on data that is subject to change. This unpredictability can lead to unexpected behavior and errors.
-
Potential Data Corruption:
- In some cases, decisions made based on dirty reads can lead to corrupt data states that are difficult to detect and correct.
Scenario: Bank Account Transfers
-
Transaction T1: User A transfers $500 from their checking account to their savings account.
- Step 1: Deduct $500 from checking account (uncommitted).
- Step 2: Add $500 to savings account (uncommitted).
-
Transaction T2: User A checks their total balance across both accounts.
- Reads checking account balance.
- Reads savings account balance.
Problem:
- Dirty Read: Transaction T2 reads the balance after the deduction but before the addition.
- Checking account balance: $500 (after deduction).
- Savings account balance: $1000 (before addition).
- Total balance seen by the user: $500 + $1000 = $1500.
- T1 is rolled back due to an error.
- Actual balance should remain $1000 + $1000 = $2000.
- User sees an incorrect total balance of $1500, leading to confusion and possible incorrect financial decisions.
Scenario: E-commerce Inventory System
-
Transaction T1: An item is sold, reducing inventory.
- Step 1: Reduce item inventory by 1 (uncommitted).
- Step 2: Confirm sale and commit changes.
-
Transaction T2: A customer checks the availability of the item.
- Reads the current inventory level.
Problem:
- Dirty Read: Transaction T2 reads the inventory level after the reduction but before the transaction is committed.
- Inventory before sale: 10 units.
- Inventory after uncommitted reduction: 9 units.
- T1 fails and rolls back, keeping the inventory at 10 units.
- T2 reports 9 units available.
- Customer places an order based on incorrect inventory information, potentially leading to overselling and customer dissatisfaction.
Scenario: Real-Time Data Analytics
-
Transaction T1: Data from multiple sensors is aggregated and temporarily stored for real-time analytics.
- Step 1: Insert sensor data (uncommitted).
- Step 2: Calculate aggregate values (uncommitted).
-
Transaction T2: A report is generated based on the aggregated data.
- Reads aggregated data.
Problem:
- Dirty Read: Transaction T2 reads the aggregated data that includes uncommitted sensor data.
- Sensor data before aggregation: 50 readings.
- Aggregated result after uncommitted inserts: 60 readings.
- T1 fails, and the uncommitted sensor data is rolled back.
- Report generated shows incorrect aggregated data, leading to inaccurate analytics and potentially flawed business decisions.
To avoid dirty reads, databases can employ higher isolation levels that prevent transactions from reading uncommitted data. Common isolation levels that prevent dirty reads include:
-
Read Committed:
- Transactions can only read data that has been committed by other transactions. This isolation level ensures that dirty reads do not occur.
-
Repeatable Read:
- Ensures that if a transaction reads a row, it will see the same data if it reads that row again within the same transaction, preventing non-repeatable reads as well as dirty reads.
-
Serializable:
- Provides the strictest level of isolation, ensuring that transactions are completely isolated from each other and preventing all types of read anomalies, including dirty reads, non-repeatable reads, and phantom reads.
Dirty reads can lead to various issues, including data inconsistencies, unreliable applications, cascading rollbacks, and potential data corruption. By using appropriate isolation levels, databases can prevent dirty reads and ensure data integrity and consistency.
What is the downside of using the same row level locks to prevent dirty reads? What is the common approach used in practice
Long running write transaction can force read only transactions to wait until that long running transaction has completed.
Commonly used approach is that databases remember both old committed value and new value by the transaction that holds the lock. Read requests get the old value so that writer does not block reader.
DDIA 237
How do we deal with dead row level locks?
Database automatically detect deadlocks and abort one of them so that the others can make progress. The aborted transaction needs to be retried by the application.
DDIA 258
What does it mean that a database implements read committed isolation
A DB that blocks both dirty writes and dirty reads.
DDIA 236
What is read skew? Give an example
Also called non repeatable read.
We are reading committed data but there is a write executed in the middle of massive read, and therefore some invariant doesn’t hold anymore.
Long analytical read.
DDIA 238
How do we resolve a read skew problem
Snapshot isolation - DDIA 238
Assign a monotonic increasing number to each write. Store all the data overridden with a transaction number.
The transaction sees all the data that was committed at the start of the transaction. Subsequent transactions shouldn’t impact.
A key principle is writers never block readers and vice versa.
What is lost updates
Increment the counter.
Read-modify-write.
Atomic write operations.
DDIA 243
———————
A lost update occurs when two transactions read the same data and then both attempt to update it. The changes made by the first transaction are overwritten by the second transaction, resulting in a loss of the first update. This anomaly typically happens in a scenario where two transactions are trying to write to the same data item without proper synchronization.
- Transaction 1 (T1) reads a value of 10 from a row (e.g., account balance).
- Transaction 2 (T2) reads the same value of 10 from the same row.
- T1 updates the value to 15 and commits.
- T2 updates the value to 20 and commits.
- The final value is 20, and the update made by T1 is lost.
Lost updates are not inherently prevented by read committed isolation. It is handled by repeatable reads isolation with 2 phase locking.
What is write skew and how to prevent
Database invariant is broken due to changes to different rows.
Solution - hold the locks for all rows that can impact the same database variant.
DDIA 247
What is phantoms writes and how to prevent
Two writes that create new rows which conflict with each other.
A write in one transaction changes the results of a search query in another transaction. There is no object to which we can attach the locks.
Materializing conflicts
DDIA 251
What is serializability in database
When transactions are committed, the result is the same as if they had run serially, even though in reality they may have run concurrently.
Serializability in databases ensures that regardless of how transactions are interleaved in their execution, their effect on the database state is as if they had been executed in some serial order, thus preserving the consistency and correctness of the data. This level of isolation is critical for applications requiring strong consistency guarantees, such as financial services, where transaction integrity is paramount.
DDIA 225
What is actual serial execution and why do we use it
Feasible since 2007.
Runs everything in one core with one thread sequentially. After 2007 it’s feasible because RAM became cheap enough and sometimes we can fit entire dataset in memory.
OLTP transactions are usually short and only make a small number of reads and writes Long running OLAP queries can run on consistent snapshot using snapshot isolation outside of the serial execution loop.
Sometimes single threaded execution runs faster than a system that supports concurrency because it get rid of the locking overhead. However the throughout is limited to a single CPU and therefore not scalable.
VoltDB/H-store
DDIA 253
What is stored procedures and why do we use it? What are the shortcomings with it?
Part of SQL standard since 1999.
To minimize the human and network call overhead.
System with single threaded serial transaction processing don’t allow interactive multi-statement transactions (multiple http requests).
The application must submit the entire transaction code to the database ahead of time, as a stored procedure. Hard to version control and deploy the code.
DDIA 254
What is two phase locking
MySQL. Since 1970s.
Why do we need it? To handle concurrent reads and write to the same piece of data （even in a single node system) so we don’t have dirty reads, dirty writes, Phantom reads or writes.
Read locks are shared but write locks are exclusive. Writer basically blocks all other writers and readers.
Locks are ALWAYS held until transaction commits.
It was the only widely used algorithm for serializability for around 30 yrs, but a big downside is performance.
———————
Two-phase locking (2PL) is a concurrency control method used to ensure serializability, the highest level of isolation in database systems.
Serializability ensures that the execution of transactions is equivalent to some serial order, thereby preventing all types of read and write anomalies, including dirty reads, non-repeatable reads, and phantom reads.
TPL consists of two phases: the growing phase (lock acquisition) and the shrinking phase (lock release).
Phantom reads occur when a transaction retrieves a set of records that satisfy a condition, but during the course of the transaction, another transaction inserts or deletes records that would also satisfy that condition, causing the first transaction to see a different set of records upon re-execution.
Here’s how two-phase locking prevents phantom reads:
- Locks on Read Operation: When a transaction reads a set of records that satisfy a condition, two-phase locking ensures that the transaction acquires shared locks (read locks) on all records that satisfy the condition. These locks prevent other transactions from inserting or deleting records that would affect the set of records being read by the current transaction.
- Locks on Write Operation: When a transaction inserts or deletes records, it acquires exclusive locks (write locks) on those records. These locks prevent other transactions from reading or writing to the affected records until the transaction holding the locks completes and releases them.
By ensuring that transactions acquire appropriate locks (shared or exclusive) on all records they access or modify during the growing phase and release those locks only during the shrinking phase, two-phase locking prevents other transactions from inserting or deleting records that would cause phantom reads.
In essence, two-phase locking guarantees that the set of records a transaction reads remains unchanged throughout its execution, thereby preventing the occurrence of phantom reads.
Yes, in the context of preventing phantom reads, two-phase locking (2PL) does involve the use of predicate locks, though the term “predicate lock” itself is more commonly associated with other concurrency control mechanisms like predicate-based locking.
Here’s how two-phase locking relates to predicate locks in preventing phantom reads:
- Predicate Locks in Two-Phase Locking: In two-phase locking, transactions acquire locks on individual data items (records) either in a shared (read) mode or exclusive (write) mode. When a transaction performs a read operation with a condition (predicate), such as a WHERE clause in SQL, it acquires shared locks on all records that satisfy the condition. These shared locks effectively act as predicate locks because they protect the set of records that satisfy the condition from being modified (inserted or deleted) by other transactions.
- Preventing Phantom Reads: Phantom reads occur when a transaction re-executes a query and finds a different set of records due to insertions or deletions by other transactions. By acquiring shared locks on all records that match a predicate during the read operation, two-phase locking ensures that the set of records remains consistent for the duration of the transaction. This prevents other transactions from inserting or deleting records that would affect the results of the current transaction’s query, thereby preventing phantom reads.
In summary, while two-phase locking is not typically described using the term “predicate locks” in the same way as other concurrency control mechanisms, the concept of acquiring shared locks on records that satisfy a predicate condition effectively serves the purpose of preventing phantom reads by ensuring consistent query results across transaction executions.
———————
DDIA 257
What is predicate locks
Rather than a lock on an object, it is a lock on patterns, on certain query conditions.
Pretty slow to run to evaluate the full query.
If two phase locking include predicate locks, the database prevent all forms of write skew and other race conditions, so its isolation becomes serializable.
DDIA 259
What is index range locks
Simplify the predicate lock by matching a greater set of objects to improve the performance.
DDIA 259
What is database snapshot
A consistent state of a database at some point of time.
What’s serializable snapshot isolation and when should we use it
Introduced in 2008. Used by postgreSQL; Oracle; Microsoft SQL.
The optimistic concurrency control.
Transactions continue anyway in the hope that everything will turn out all right. When it wants to commit, the database checks whether anything bad happened. Only transaction that executed serializably are allowed to commit.
Only when most transaction do not get into the way of each other. Don’t use it if two many conflicting transactions.
Column oriented storage
Definition
Use case
Parquet is a free and open source column oriented storage format that supports a document data model, since 2013.
The idea is pretty simple - don’t store the values from one row together, stores all the values from each column together instead. Query only needs to read and parse those columns that are used in that query, which can save a lot of work and minimize the disk I/O.
We can perform column compaction to reduce the amount of data we store; we can even store and cache those data in memory to speed up query.
Downside - every write needs to go to many different places on disk.
Can write rows into LSM trees in memory and flush them all to disk and do bulk writes basically column oriented files.
Column storage - bitmap compression + run-length encoding
Often, the number of distinct values in a column is small compared to the number of rows (for example, a retailer may have billions of sales transactions, but only 100,000 distinct products). We can now take a column with n distinct values and turn it into n separate bitmaps: one bitmap for each distinct value, with one bit for each row. The bit is 1 if the row has that value, and 0 if not.
If n is very small (for example, a country column may have approximately 200 distinct values), those bitmaps can be stored with one bit per row. But if n is bigger, there will be a lot of zeros in most of the bitmaps (we say that they are sparse). In that case, the bitmaps can additionally be run-length encoded, as shown at the bottom of the figure below. This can make the encoding of a column remarkably compact.
Bitmap indexes such as these are very well suited for the kinds of queries that are common in a data warehouse. For example:
WHERE product_sk IN (30, 68, 69):
Load the three bitmaps for product_sk = 30, product_sk = 68, and product_sk = 69, and calculate the bitwise OR of the three bitmaps, which can be done very efficiently.
Predicate push down
A db optimization technique that improves query performance by filtering data before it’s transferred.
Entire dataset don’t need to be scanned during query execution.
Example
Consider a database query intended to find all transactions over $1,000 in a financial database:
SELECT * FROM transactions WHERE amount > 1000;
With predicate pushdown, the database engine would apply the filter amount > 1000 directly when reading the data from disk, rather than fetching all transactions into memory and then filtering them.
Conclusion
Predicate pushdown is a critical optimization in the management and processing of data across various platforms, from traditional RDBMS to modern big data systems. By ensuring that unnecessary data is filtered out early in the data processing pipeline, systems can achieve higher performance and efficiency.
What are the downsides of column oriented storage?
Sorted.
Writes to many columns.
Use LSM tree to do bulk update.
What is forward and backward compatibility
Forward: old code can read records that were written by new code.
Backward - new code can read records written by old code. If you were to add a field and make it required, that check would fail if new code read data written by old code, because the old code will not have written the new field that you added. Therefore, to maintain backward compatibility, every field you add after the initial deployment of the schema must be optional or have a default value (automatic backfill or manual backfill)
Backward compatibility is normally not hard to achieve: as author of the newer code, you know the format of data written by older code, and so you can explicitly handle it (if necessary by simply keeping the old code to read the old data). Forward compatibility can be trickier, because it requires older code to ignore additions made by a newer version of the code.
DDIA 121
What are Apache Thrift and Protocol buffers
Apache Thrift and Protocol Buffers (protobuf) are binary encoding libraries that are based on the same principle. Protocol Buffers was originally developed at Google, Thrift was originally developed at Facebook, and both were made open source in 2007–08.
Both Thrift and Protocol Buffers require a schema for any data that is encode. To encode the data in Thrift, you would describe the schema in the
Thrift interface definition language (IDL) like this:
struct Person {
1: required string userName,
2: optional i64 favoriteNumber,
3: optional list<string> interests</string>
}
Each field has a type annotation (to indicate whether it is a string, integer, list, etc.)
and, where required, a length indication (length of a string, number of items in a list).
The strings that appear in the data (“Martin”, “daydreaming”, “hacking”) are also encoded as ASCII (or rather, UTF-8).
DDIA 117
What’s the disadvantages of thrift or protocol buffers
Require knowledge of scheme at compile time.
What is Apache avro
Apache Avro is another binary encoding format that is interestingly different from Protocol Buffers and Thrift. It was started in 2009 as a subproject of Hadoop, as a result of Thrift not being a good fit for Hadoop’s use cases.
Create database schema on the fly based off of column names.
Binary data can only be decoded correctly if the code reading the data is using the exact same schema as the code that wrote the data.
To parse the binary data, you go through the fields in the order that they appear in the schema and use the schema to tell you the datatype of each field.
The key idea with Avro is that the writer’s schema and the reader’s schema don’t have to be the same—they only need to be compatible. When data is decoded (read), the Avro library resolves the differences by looking at the writer’s schema and the reader’s schema side by side and translating the data from the writer’s schema into the reader’s schema.
For example, it’s no problem if the writer’s schema and the reader’s schema have their fields in a different order, because the schema resolution matches up the fields by field name. If the code reading the data encounters a field that appears in the writer’s schema but not in the reader’s schema, it is ignored. If the code reading the data expects some field, but the writer’s schema does not contain a field of that name, it is filled in with a default value declared in the reader’s schema.
DDIA 122
Why do we need replication
Replication means keeping a copy of the same data on multiple machines that are connected via a network. As discussed in the introduction to Part II, there are several reasons why you might want to replicate data:
• To keep data geographically close to your users (and thus reduce latency)
• To allow the system to continue working even if some of its parts have failed (and thus increase availability)
• To scale out the number of machines that can serve read queries (and thus increase read throughput)
Page 151
What is synchronous and asynchronous replication
The advantage of synchronous replication is that the follower is guaranteed to have an up-to-date copy of the data that is consistent with the leader. If the leader suddenly fails, we can be sure that the data is still available on the follower.
The disadvantage is that if the synchronous follower doesn’t respond (because it has crashed, or there is a network fault, or for any other reason), the write cannot be processed. The leader must block all writes and wait until the synchronous replica is available again.
For that reason, it is impractical for all followers to be synchronous: any one node outage would cause the whole system to grind to a halt. In practice, if you enable synchronous replication on a database, it usually means that one of the followers is synchronous, and the others are asynchronous.
If the synchronous follower becomes unavailable or slow, one of the asynchronous followers is made synchronous. This guarantees that you have an up-to-date copy of the data on at least two nodes: the leader and one synchronous follower. This configuration is sometimes also called semi-synchronous.
Often, leader-based replication is configured to be completely asynchronous. In this case, if the leader fails and is not recoverable, any writes that have not yet been replicated to followers are lost. This means that a write is not guaranteed to be durable, even if it has been confirmed to the client.
However, a fully asynchronous configuration has the advantage that the leader can continue processing writes, even if all of its followers have fallen behind. Weakening durability may sound like a bad trade-off, but asynchronous replication is nevertheless widely used, especially if there are many followers or if they are geographically distributed.
DDIA 153
How do we implement replication?
Logical (row-based) log replication
An alternative is to use different log formats for replication and for the storage engine, which allows the replication log to be decoupled from the storage engine internals.
This kind of replication log is called a logical log, to distinguish it from the storage engine’s (physical) data representation.
A logical log for a relational database is usually a sequence of records describing writes to database tables at the granularity of a row:
• For an inserted row, the log contains the new values of all columns.
• For a deleted row, the log contains enough information to uniquely identify the row that was deleted. Typically this would be the primary key, but if there is no primary key on the table, the old values of all columns need to be logged.
• For an updated row, the log contains enough information to uniquely identify the updated row, and the new values of all columns (or at least the new values of all columns that changed).
You know it. Find photos.
DDIA 159
What’s the issue with WAL as replication log?
The main disadvantage is that the log describes the data on a very low level: a WAL contains details of which bytes were changed in which disk blocks. This makes replication closely coupled to the storage engine. If the database changes its storage format from one version to another, it is typically not possible to run different versions of the database software on the leader and the followers.
That may seem like a minor implementation detail, but it can have a big operational impact. If the replication protocol allows the follower to use a newer software version than the leader, you can perform a zero-downtime upgrade of the database software by first upgrading the followers and then performing a failover to make one of the upgraded nodes the new leader. If the replication protocol does not allow this version mismatch, as is often the case with WAL shipping, such upgrades require downtime.
How do we implement read-after-write consistency
How can we implement read-after-write consistency in a system with leader-based replication? There are various possible techniques. To mention a few:
• When reading something that the user may have modified, read it from the leader; otherwise, read it from a follower. This requires that you have some way of knowing whether something might have been modified, without actually querying it. For example, user profile information on a social network is normally only editable by the owner of the profile, not by anybody else. Thus, a simple rule is: always read the user’s own profile from the leader, and any other users’ profiles from a follower.
• If most things in the application are potentially editable by the user, that approach won’t be effective, as most things would have to be read from the leader (negating the benefit of read scaling). In that case, other criteria may be used to decide whether to read from the leader. For example, you could track the time of the last update and, for one minute after the last update, make all reads from the leader. You could also monitor the replication lag on followers and prevent queries on any follower that is more than one minute behind the leader.
• The client can remember the timestamp of its most recent write—then the system can ensure that the replica serving any reads for that user reflects updates at least until that timestamp. If a replica is not sufficiently up to date, either the read can be handled by another replica or the query can wait until the replica has caught up. The timestamp could be a logical timestamp (something that indicates ordering of writes, such as the log sequence number) or the actual system clock
Monotonic reads and what’s the edge case it is preventing?
In the context of single leader and asynchronous replication; client may see things move backwards in time.
Monotonic reads is a guarantee that this kind of anomaly does not happen. It’s a lesser guarantee than strong consistency, but a stronger guarantee than eventual consistency. When you read data, you may see an old value; monotonic reads only means that if one user makes several reads in sequence, they will not see time go backward— i.e., they will not read older data after having previously read newer data.
One way of achieving monotonic reads is to make sure that each user always makes their reads from the same replica (different users can read from different replicas). For example, the replica can be chosen based on a hash of the user ID, rather than randomly. However, if that replica fails, the user’s queries will need to be rerouted to another replica.
DDIA 165
Give an example of violation of causality.
DDIA 165
What is consistent prefix reads? How is it implemented?
All casually dependent writes goes to same replica.
Preventing this kind of anomaly requires another type of guarantee: consistent prefix reads. This guarantee says that if a sequence of writes happens in a certain order, then anyone reading those writes will see them appear in the same order. This is a particular problem in partitioned (sharded) databases. If the database always applies writes in the same order, reads always see a consistent prefix, so this anomaly cannot happen. However, in many distributed databases, different partitions operate independently, so there is no global ordering of writes: when a user reads from the database, they may see some parts of the database in an older state and some in a newer state. One solution is to make sure that any writes that are causally related to each other are written to the same partition—but in some applications that cannot be done efficiently.
DDIA 166
What’s the challenges with leader failure?
- How to detect leader is down - timeout settings
- Lost write - leader goes down before new writes get propagated to any of the followers
- Split brain
The topologies for multi leaders sync
DDIA 175
What is the problem of leaderless replication via gossip protocol
- It is expensive to ensure causality using lamport clock or detecting concurrent writes using version vectors and resolve conflicts.
- Infinite loop of broadcasting messages - for a specific write, maintain a set of visited nodes
DDIA 177
Last write wins to resolve write conflicts
DDIA 186
What timestamp we can use?
Quartz crystal vibration.
Clock skew.
NTP network time protocol - GPS clock
Adjust computer clock
What is concurrent writes in multi-leader system
The writes don’t know about each other
How to detect concurrent write in multi-leader system
Version vector - number of writes it has seen from each leader node
What are the other solutions besides LWW to mitigate concurrent writes problem
One option is to store sibling and let user decide later. As merging siblings in application code is complex and error-prone, there are some efforts to design data structures that can perform this merging automatically. For example, Riak’s datatype support uses a family of data structures called CRDTs that can automatically merge siblings in sensible ways, including preserving deletions.
CRDT - conflict free replicated data types, formally introduced in 2011 and now used by Riak, Redis l, AntidoteDB.
What is operational CRDT
When we say that “updates are propagated as operations” in the context of operation-based CRDTs (CmRDTs), we’re referring to the method by which changes to data are communicated across different nodes in a distributed system.
Here’s a detailed breakdown of what this entails: Process of Propagating Updates as Operations
1. Operation Encapsulation: Each change made to the data is encapsulated as a discrete operation. This operation includes all the information necessary to replicate the change on other nodes. For instance, if you’re incrementing a counter, the operation might specify that the counter should be incremented by one. 2. Transmission: Instead of sending the entire state of the data structure (which could be large), only the operation itself is sent across the network to other nodes. This is typically more efficient, especially when the data structure is large but the changes are small. 3. Commutative Operations: Operations are designed to be commutative. This means that the order in which operations are applied does not affect the final outcome. For example, if two nodes simultaneously increment the same counter, the operations “increment by 1” can be applied in any order but still yield the same result.
Example
Consider a collaborative text editing application where multiple users can insert or delete characters in a document:
• User A inserts the letter “a” at position 1. • User B deletes the character at position 2.
Each of these actions (insertion and deletion) is an operation. These operations are sent to all other participating nodes:
• The insertion operation from User A might be encoded as insert('a', 1).
• The deletion operation from User B might be encoded as delete(2).
When another node receives these operations, it applies them to its local version of the document. Because the operations are commutative and designed to be independent of specific states, they can be applied in any order without resulting in conflicts or inconsistencies.
Benefits
This approach minimizes the amount of data transferred between nodes, reduces latency in response times, and ensures that even if messages arrive out of order or are delayed, the final state of the data structure remains consistent across all nodes. This makes operation-based CRDTs highly effective for real-time, distributed applications where low latency and high reliability are crucial.
However, whenever we have casual relationships amongst our operations and operational CRDT could let us down.
Stated based CRDT
Updates are managed by sharing the entire state of the data structure rather than individual operations.
Each node independently updates its local state, and periodically, the entire state is merged with states from other nodes using a merge operation that is associative, commutative, and idempotent. This ensures that all nodes eventually converge to the same state, regardless of the order in which states are merged.
A G-Counter is a simple CRDT used to implement a distributed counter that can only increase.
- Local Update: Each node maintains its own counter and increments its local count independently without coordination with other nodes.
- Merging: When nodes communicate, they share their entire counter state. The merge operation for a G-Counter takes the maximum of each node’s count for each node in the system, effectively synchronizing the counts across all nodes to the highest value observed.
- Result: The result is a counter that always increases and never decreases, providing a reliable count even in the presence of network partitions and asynchronous updates.
State-based CRDTs like the G-Counter are particularly useful in distributed systems where it’s important to ensure that all nodes can operate independently and still maintain consistency across the system without complex conflict resolution mechanisms.
State-based CRDTs handle causal relationships by ensuring that the merge operation can combine states from different nodes in a way that reflects the causal order of updates, even though individual nodes might apply updates independently and asynchronously. For operations and data structures where the relationship between updates is not inherently commutative, like a G-Counter, CRDTs must incorporate additional mechanisms to track and reconcile the order of operations. One common approach is to use version vectors or other metadata that capture the causal relationships between updates.
What database use leaderless replication
Cassandra, Riak
What is the context of read repair?
Read repair is a technique used in distributed data storage systems to ensure data consistency across different replicas of the same data. This process is particularly important in systems that employ replication to provide high availability and fault tolerance. Read repair helps in keeping the replicas synchronized by actively correcting discrepancies during read operations.
Read repair operates under eventual consistency models, where data is replicated across multiple nodes to ensure availability and resilience against node failures. Here’s how read repair typically functions:
- Data Retrieval: When a read request is issued to a distributed database, it may retrieve data from one or more replicas, depending on the consistency requirements specified for the read operation.
- Version Comparison: The system compares the versions of the data retrieved from different replicas. This can be done by checking timestamps, version numbers, or other metadata associated with the data to determine which replica has the most recent version.
- Discrepancy Detection: If discrepancies are found (i.e., some replicas have older versions of the data), the system identifies the most up-to-date version of the data.
- Correction: The system then propagates the correct version back to those replicas that had outdated or incorrect versions, thus “repairing” the data on those nodes.
- Update of Replicas: The outdated replicas update their data to reflect the most recent version, ensuring consistency across the system.
Read repair can be triggered in a few different ways, depending on the system’s design and configuration:
- Synchronous Read Repair: This happens as part of the read process. The system checks and repairs any inconsistencies before returning the data to the requester. This ensures strong consistency but can increase the latency of read operations because the system must wait until all necessary repairs are made.
- Asynchronous Read Repair: In this approach, the system logs the found discrepancies during the read operation but does not immediately correct them. Instead, the repairs are done in the background, allowing the read operation to complete with lower latency. This method favors availability and performance over immediate consistency.
Read repair is widely used in various distributed data stores, including:
- Cassandra: Uses read repair as part of its normal operations to ensure that all replicas for a given piece of data remain consistent over time. Cassandra allows configuring the probability of read repair happening on a per-read basis.
- DynamoDB: Amazon’s DynamoDB uses a form of read repair in its background processes to maintain data consistency across its distributed storage.
- Riak: Also implements read repair to handle discrepancies found during read operations, ensuring that data returned to the user is the most current and that all replicas are eventually consistent.
Read repair is a crucial mechanism in distributed databases to maintain data consistency across replicas. By correcting discrepancies during read operations, read repair helps ensure that all replicas converge towards a consistent state, thereby fulfilling the promises of reliability and availability in distributed systems. This technique is particularly valuable in environments where data is continuously updated and accessed across multiple nodes.
What is the context of anti-entropy and how is it implemented
Anti-entropy is a concept used in distributed systems to ensure data consistency across replicas. It involves mechanisms that detect and correct discrepancies in data at different nodes, effectively reducing “entropy” which in this context refers to disorder or inconsistencies in the system. Anti-entropy processes help in synchronizing data across a distributed network by continuously or periodically comparing data sets and resolving differences.
In distributed data storage systems, replicas of the same data may become inconsistent due to network failures, temporary partitioning, or delays in update propagation. Anti-entropy is a proactive approach to handle these potential inconsistencies and ensure that all replicas converge to the same state, even in the face of failures.
-
Merkle Trees:
- Concept: A Merkle tree is a binary tree in which every leaf node is labelled with the hash of a data block, and every non-leaf node is labelled with the cryptographic hash of the labels of its child nodes. Merkle trees are used to efficiently compare data sets across different nodes.
- Usage: When two nodes need to synchronize their data, they can exchange the root hashes of their Merkle trees. If the hashes differ, they can compare smaller segments of their trees (i.e., branches) progressively, until they localize the blocks that differ. This process significantly reduces the amount of data that needs to be transferred for synchronization.
- Example: Cassandra uses Merkle trees during its anti-entropy repair process to identify and synchronize differences between replicas.
-
Read Repair and Write Repair:
- Read Repair: As previously explained, read repair involves detecting and correcting inconsistencies during read operations. This is a form of anti-entropy because it actively seeks to eliminate divergences by updating replicas with the most recent data version when a discrepancy is detected during a read.
- Write Repair: Similar to read repair, but corrections are made during write operations. When new data is written or existing data is updated, the system ensures that all replicas are immediately brought up to date, hence maintaining consistency.
-
Gossip Protocols:
- Concept: Gossip protocols involve nodes periodically sharing state information about themselves and other nodes they know about in a manner akin to how gossip spreads in social networks.
- Usage: Each node in a distributed system periodically exchanges state information with a randomly chosen other node. This state information typically includes data about the node’s view of the system and can include version vectors or other metadata that track data state.
- Purpose: The continuous, random mixing of state information helps in quickly propagating updates across all nodes and correcting any discrepancies, thereby maintaining system consistency over time.
- Data Consistency: Ensures all nodes in a distributed system eventually hold the same data, despite any failures or inconsistencies.
- Fault Tolerance: Increases the resilience of the system to network failures, data corruption, or other anomalies.
- Scalability: Allows the system to scale effectively, as the mechanisms can handle the synchronization of data across a large number of nodes without central coordination.
Anti-entropy mechanisms are essential in distributed systems to manage data consistency proactively. By continuously addressing and correcting data discrepancies, these mechanisms help ensure that the system operates reliably and efficiently, even as it scales and faces various operational challenges.
Quorum writes and reads
You know it
What is the edge case that quorum consistency may fail
If a sloppy quorum is used, the w writes may end up on different nodes than the r reads, so there is no longer a guaranteed overlap between the r nodes and the w nodes.
Still prone to concurrent writes until all nodes reach distributed consensus.
Writes failed at some nodes.
What is strong consistency?
Everyone reading the same time would agree on an identical and most up to date value for a given key
What is sloppy quorum and hinted handoff
In a large cluster (with significantly more than n nodes) it’s likely that the client can connect to some database nodes during the network interruption, just not to the nodes that it needs to assemble a quorum for a particular value.
In that case, database designers face a trade-off:
• Is it better to return errors to all requests for which we cannot reach a quorum of w or r nodes?
• Or should we accept writes anyway, and write them to some nodes that are reachable but aren’t among the n nodes on which the value usually lives?
The latter is known as a sloppy quorum: writes and reads still require w and r successful responses, but those may include nodes that are not among the designated n “home” nodes for a value.
By analogy, if you lock yourself out of your house, you may knock on the neighbor’s door and ask whether you may stay on their couch temporarily. Once the network interruption is fixed, any writes that one node temporarily accepted on behalf of another node are sent to the appropriate “home” nodes. This is called hinted handoff. (Once you find the keys to your house again, your neighbor politely asks you to get off their couch and go home.)
Thus, a sloppy quorum actually isn’t a quorum at all in the traditional sense. It’s only an assurance of durability, namely that the data is stored on w nodes somewhere. There is no guarantee that a read of r nodes will see it until the hinted handoff has completed.
Sloppy quorums are optional in all common Dynamo implementations. In Riak they are enabled by default, and in Cassandra and Voldemort they are disabled by default
Why do we need database partitioning
- Data volume
- Scalability - large dataset can be distributed across many disks and query load can be distributed across many processors.
Replication is having multiple copies of the same data on different nodes. For very large datasets, or very high query throughput, that is not sufficient: we need to break the data up into partitions, also known as sharding.
The main reason for wanting to partition data is scalability. Different partitions can be placed on different nodes in a shared-nothing cluster, and the query load can be distributed across many processors.
Partitioned databases were pioneered in the 1980s by products such as Teradata and Tandem NonStop SQL, and more recently rediscovered by NoSQL databases and Hadoop-based data warehouses.
Some systems are designed for transactional workloads, and others for analytics; this difference affects how the system is tuned, but the fundamentals of partitioning apply to both kinds of workloads.
DDIA 199
Range based partitioning; pros and cons
Database locality for range scan;
Hot spot - last name starting with x or z: a to c partition.
If you know the boundaries between the ranges, you can easily determine which partition contains a given key. If you also know which partition is assigned to which node, then you can make your request directly to the appropriate node (or, in the case of the encyclopedia, pick the correct book off the shelf).
However, the downside of key range partitioning is that certain access patterns can lead to hot spots. If the key is a timestamp, then the partitions correspond to ranges of time—e.g., one partition per day. Unfortunately, because we write data from the sensors to the database as the measurements happen, all the writes end up going to the same partition (the one for today), so that partition can be overloaded with writes while others sit idle.
To avoid this problem in the sensor database, you need to use something other than the timestamp as the first element of the key. For example, you could prefix each timestamp with the sensor name so that the partitioning is first by sensor name and then by time. Assuming you have many sensors active at the same time, the write load will end up more evenly spread across the partitions. Now, when you want to fetch the values of multiple sensors within a time range, you need to perform a separate range query for each sensor name.
DDIA 202
Hash range based partitioning
Each partition assigned a range of hash keys.
Even distribution of keys, less hot spot
No data locality for range queries
Keys that were once adjacent are now scattered across all the partitions, so their sort order is lost. In MongoDB, if you have enabled hash-based sharding mode, any range query has to be sent to all partitions.
Range queries on the primary key are not supported by Riak, Couchbase, or Voldemort.
Cassandra achieves a compromise between the two partitioning strategies. A table in Cassandra can be declared with a compound primary key consisting of several columns.
Only the first part of that key is hashed to determine the partition, but the other columns are used as a concatenated index for sorting the data in Cassandra’s SSTables. A query therefore cannot search for a range of values within the first column of a compound key, but if it specifies a fixed value for the first column, it can perform an efficient range scan over the other columns of the key.
Bottom of DDIA 203
Why is hash range partitioning still not perfectly solving hot spot
If certain keys are extremely active. One celebrity has too many comments for her instagram posts.
Today, most data systems are not able to automatically compensate for such a highly skewed workload, so it’s the responsibility of the application to reduce the skew.
For example, if one key is known to be very hot, a simple technique is to add a random number to the beginning or end of the key. Just a two-digit decimal random number would split the writes to the key evenly across 100 different keys, allowing those keys to be distributed to different partitions.
However, having split the writes across different keys, any reads now have to do additional work, as they have to read the data from all 100 keys and combine it.
This technique also requires additional bookkeeping: it only makes sense to append the random number for the small number of hot keys; for the vast majority of keys with low write throughput this would be unnecessary overhead. Thus, you also need some way of keeping track of which keys are being split.
Perhaps in the future, data systems will be able to automatically detect and compensate for skewed workloads; but for now, you need to think through the trade-offs for your own application.
How to optimize the hash range key partitioning for range query
Having secondary indexes.
A secondary index usually doesn’t identify a record uniquely but rather is a way of searching for occurrences of a particular value: find all actions by user 123, find all articles containing the word hogwash, find all cars whose color is red, and so on.
Local secondary indexes
Second copy of sorted dataset on each partition. Pro: No need to perform any extra writes over the network. Con: Need to fetch from every single partition.
In this indexing approach, each partition is completely separate: each partition maintains its own secondary indexes, covering only the documents in that partition. It doesn’t care what data is stored in other partitions.
Whenever you need to write to the database—to add, remove, or update a document—you only need to deal with the partition that contains the document ID that you are writing. For that reason, a document-partitioned index is also known as a local index (as opposed to a global index, described in the next section).
However, reading from a document-partitioned index requires care: unless you have done something special with the document IDs, there is no reason why all the cars with a particular color or a particular make would be in the same partition. Thus, if you want to search for red cars, you need to send the query to all partitions, and combine all the results you get back. This approach to querying a partitioned database is sometimes known as scatter/ gather, and it can make read queries on secondary indexes quite expensive. Even if you query the partitions in parallel, scatter/gather is prone to tail latency amplification (see “Percentiles in Practice” on page 16).
Nevertheless, it is widely used: MongoDB, Riak, Cassandra, Elasticsearch, SolrCloud, and VoltDB all use document-partitioned secondary indexes.
Most database vendors recommend that you structure your partitioning scheme so that secondary index queries can be served from a single partition, but that is not always possible, especially when you’re using multiple secondary indexes in a single query (such as filtering cars by color and by make at the same time).
Global secondary indexes
Take the aggregated dataset across partitions and we split that into pieces which are stored across nodes.
A global index must also be partitioned, but it can be partitioned differently from the primary key index.
The advantage of a global (term-partitioned) index over a document-partitioned index is that it can make reads more efficient: rather than doing scatter/gather over all partitions, a client only needs to make a request to the partition containing the term that it wants. However, the downside of a global index is that writes are slower and more complicated, because a write to a single document may now affect multiple partitions of the index (every term in the document might be on a different partition, on a different node). In an ideal world, the index would always be up to date, and every document written to the database would immediately be reflected in the index. However, in a termpartitioned index, that would require a distributed transaction across all partitions affected by a write, which is not supported in all databases. In practice, updates to global secondary indexes are often asynchronous (that is, if you read the index shortly after a write, the change you just made may not yet be reflected in the index).
For example, Amazon DynamoDB states that its global secondary indexes are updated within a fraction of a second in normal circumstances, but may experience longer propagation delays in cases of faults in the infrastructure.
How is two phase commit implemented
Coordinator; prepare request and track responses from all participants; commit or abort requests
2PC uses a new component that does not normally appear in single-node transactions: a coordinator (also known as transaction manager). The coordinator is often implemented as a library within the same application process that is requesting the transaction (e.g., embedded in a Java EE container), but it can also be a separate process or service. Examples of such coordinators include Narayana, JOTM, BTM, or MSDTC.
A 2PC transaction begins with the application reading and writing data on multiple database nodes, as normal. We call these database nodes participants in the transaction. When the application is ready to commit, the coordinator begins phase 1: it sends a prepare request to each of the nodes, asking them whether they are able to commit.
The coordinator then tracks the responses from the participants:
• If all participants reply “yes,” indicating they are ready to commit, then the coordinator persists a new entry into its commit log, then sends out a commit request in phase 2, and the commit actually takes place.
• If any of the participants replies “no,” the coordinator sends an abort request to all nodes in phase 2.
This process is somewhat like the traditional marriage ceremony in Western cultures: the minister asks the bride and groom individually whether each wants to marry the other, and typically receives the answer “I do” from both. After receiving both acknowledgments, the minister pronounces the couple husband and wife: the transaction is committed, and the happy fact is broadcast to all attendees. If either bride or groom does not say “yes,” the ceremony is aborted.
DDIA 356
Cons of two phase commits
Coordinator failures;
Participants failures; If this request fails or times out, the coordinator must retry forever until it succeeds. There is no more going back: if the decision was to commit, that decision must be enforced, no matter how many retries it takes. If a participant has crashed in the meantime, the transaction will be committed when it recovers—since the participant voted “yes,” it cannot refuse to commit when it recovers.
Why do we need consistent hashing
Minimize data movement over network. Great for partitioning and load balancing.
Consistent hashing is a distributed hashing scheme that operates independently of the number of servers or objects in a distributed hash table by assigning them a position on an abstract circle, or hash ring.
This method is particularly useful in distributed systems because it minimizes the number of reassignments when a server is added or removed, thus providing better scalability and manageability.
In consistent hashing, both the data items (keys) and servers (nodes) are hashed onto a hash ring using a hash function. The hash function typically produces a fixed range of numerical outputs (for example, a 32-bit integer space). Here’s a step-by-step explanation of how consistent hashing works:
-
Hashing Nodes and Data:
- Each node in the system is assigned a hash value by hashing some characteristic (like its IP address or hostname). These values are treated as positions on a circular hash ring.
- Data keys are also hashed using the same hash function, mapping them to positions on the same ring.
-
Assigning Data to Nodes:
- To determine which node is responsible for a given key, the system finds the nearest node whose hash value is greater than or equal to the key’s hash. This node is often referred to as the “successor” node of the key.
- If the key’s hash is greater than the hash of any node, the key is mapped to the node with the lowest hash value due to the circular nature of the ring.
-
Handling Node Changes:
- Adding a Node: When a new node joins the network, only the keys that are mapped between the new node’s hash and the hash of its immediate predecessor node need to be reassigned to the new node.
- Removing a Node: When a node leaves or fails, its keys are reassigned to the node that immediately follows it on the hash ring.
To improve the load distribution and reassignment efficiency, consistent hashing often employs a concept called “virtual nodes”:
- Virtual Nodes (or Vnodes): Each node can be represented multiple times on the hash ring by creating multiple virtual nodes (hashes) for each server. These virtual nodes are spread out around the hash ring.
- By using virtual nodes, the system can balance the load more evenly, especially when the hash function doesn’t distribute nodes uniformly across the hash ring.
- When a physical node is added or removed, only a small proportion of keys are reassigned, because each physical node is responsible for multiple points on the ring.
Consistent hashing is widely used in various distributed systems and caching solutions, including:
- Distributed Caches: Systems like Memcached use consistent hashing to distribute cache items across multiple servers.
- Distributed Databases: NoSQL databases like Apache Cassandra and DynamoDB use consistent hashing to distribute data across multiple nodes, ensuring even data distribution and efficient node scaling.
- Scalability: It significantly reduces the need for reshuffling all keys with every change in the set of servers.
- Load Balancing: Even distribution of keys helps in load balancing across the servers.
- Fault Tolerance: Minimizes disruptions when nodes are added or removed, providing better fault tolerance.
Consistent hashing is a cornerstone technology in distributed systems, addressing key challenges related to scalability and performance. Its ability to minimize reorganization and rebalance loads efficiently makes it highly suitable for large-scale, dynamic environments.
Fixed total number of partitions
Operationally simpler.
Choosing a right number is difficult.
Too many -> overhead of management
Two few -> partition becomes large, rebalancing and recovery becomes expensive.
A fixed number of partitions is operationally simpler, and so many fixed-partition databases choose not to implement partition splitting.
Thus, the number of partitions configured at the outset is the maximum number of nodes you can have, so you need to choose it high enough to accommodate future growth. However, each partition also has management overhead, so it’s counterproductive to choose too high a number.
Choosing the right number of partitions is difficult if the total size of the dataset is highly variable (for example, if it starts small but may grow much larger over time). Since each partition contains a fixed fraction of the total data, the size of each partition grows proportionally to the total amount of data in the cluster. If partitions are very large, rebalancing and recovery from node failures become expensive. But if partitions are too small, they incur too much overhead. The best performance is achieved when the size of partitions is “just right,” neither too big nor too small, which can be hard to achieve if the number of partitions is fixed but the dataset size varies.
DDIA 211
What is linearizable databases
This is the idea behind linearizability (also known as atomic consistency, strong consistency, immediate consistency, or external consistency).
The exact definition of linearizability is quite subtle, and we will explore it in the rest of this section. But the basic idea is to make a system appear as if there were only one copy of the data, and all operations on it are atomic. With this guarantee, even though there may be multiple replicas in reality, the application does not need to worry about them.
In a linearizable system, as soon as one client successfully completes a write, all clients reading from the database must be able to see the value just written. Maintaining the illusion of a single copy of the data means guaranteeing that the value read is the most recent, up-to-date value, and doesn’t come from a stale cache or replica. In other words, linearizability is a recency guarantee. To clarify this idea, let’s look at an example of a system that is not linearizable.
DDIA 324
Lamport timestamps
DDIA 345
Every node and client keeps track of the maximum counter value it has seen so far, and includes that maximum on every write requests.
As long as the maximum counter is carried along with every operation, this scheme ensures that ordering from the lamport timestamp is consistent with causality, because casual dependency results in an increased lamport timestamp.
————————————-
There is actually a simple method for generating sequence numbers that is consistent with causality. It is called a Lamport timestamp, proposed in 1978 by Leslie Lamport, in what is now one of the most-cited papers in the field of distributed systems.
The use of Lamport timestamps is illustrated in Figure below. Each node has a unique identifier, and each node keeps a counter of the number of operations it has processed. The Lamport timestamp is then simply a pair of (counter, node ID). Two nodes may sometimes have the same counter value, but by including the node ID in the timestamp, each timestamp is made unique.
A Lamport timestamp bears no relationship to a physical time-of-day clock, but it provides total ordering: if you have two timestamps, the one with a greater counter value is the greater timestamp; if the counter values are the same, the one with the greater node ID is the greater timestamp.
The key idea about Lamport timestamps, which makes them consistent with causality, is the following: every node and every client keeps track of the maximum counter value it has seen so far, and includes that maximum on every request. When a node receives a request or response with a maximum counter value greater than its own counter value, it immediately increases its own counter to that maximum.
As long as the maximum counter value is carried along with every operation, this scheme ensures that the ordering from the Lamport timestamps is consistent with causality, because every causal dependency results in an increased timestamp.
Lamport timestamps are sometimes confused with version vectors. Although there are some similarities, they have a different purpose: version vectors can distinguish whether two operations are concurrent or whether one is causally dependent on the other, whereas Lamport timestamps always enforce a total ordering. From the total ordering of Lamport time stamps, you cannot tell whether two operations are concurrent or whether they are causally dependent. The advantage of Lamport timestamps over version vectors is that they are more compact.
The problem here is that the total order of operations only emerges after you have collected all of the operations. If another node has generated some operations, but you don’t yet know what they are, you cannot construct the final ordering of operations: the unknown operations from the other node may need to be inserted at various positions in the total order.
Raft leader election
Raft is a consensus algorithm designed for managing a replicated log across distributed systems. It’s known for its understandability and the way it handles “leader election” as part of ensuring that there’s a consistent system state across all nodes in the cluster. Here’s a detailed explanation of the steps involved in the Raft leader election process:
- All nodes start as followers: When servers start up, they begin as followers in the Raft protocol. Each node remains in follower state unless it times out (i.e., does not hear from a leader or receive valid heartbeats within a certain period).
- Timeout and candidacy: If a follower does not receive communication from the leader within a random election timeout period, it assumes there is no active leader and transitions to a candidate state to start an election. The election timeout is randomized for each follower to reduce the chances of two nodes starting their elections at the same time.
- Increment term: Upon transitioning to the candidate state, the node increments its current term (a continuously growing number representing the current election round).
- Vote for self: The candidate votes for itself first and then sends out RequestVote RPCs (remote procedure calls) to all other nodes in the cluster, asking them to vote for it.
- Majority needed: The candidate must gather a majority of votes from the cluster to become the leader. This majority ensures that there is agreement among the cluster majority.
- Handling split votes: If votes are split and no candidate wins a majority, a new election term begins, and nodes will wait for another election timeout to trigger a new election.
- Becoming a leader: If a candidate receives a majority of votes from the cluster, it transitions to the leader state. It begins to send regular heartbeats to all followers to assert its authority and prevent new elections.
- Reset if no majority: If the candidate does not receive a majority (due to a split vote or other issues), the election term ends, and nodes will reset their election timers. The process might repeat until a leader is elected.
- Heartbeats: Once a leader is established, it regularly sends heartbeat messages to all followers to maintain its authority and prevent followers from triggering new elections. These heartbeats also help in log replication processes, although they carry no log entries themselves.
- Stepping down: If, at any point, a leader or candidate discovers a higher term number (either from another node’s RPC or during communications), it immediately reverts to follower state. This behavior ensures that the node respects the newer, possibly more up-to-date authority.
This leader election process in Raft ensures that the system can recover from failures and maintain a consistent state across distributed components. The design of Raft, particularly with its strong leader and structured approach to elections and term management, makes it well-suited for a variety of distributed systems that require robust fault tolerance and data consistency.
Operational CRDT downsides
When there are casual relationships between operations and their order matters, there could be problem. We need a casually consistent broadcasting system where messages do not get dropped and they also don’t get duplicated.
Idempotent - you can do things many times and make no difference.
How does state based CRDT handles casual relationship
Handling both addition and removal operations in a state-based CRDT (CvRDT) can be complex because simple state-based data structures like Grow-Only Sets (G-Sets) or counters do not support the removal of elements directly. To manage this, a more sophisticated type of CRDT is used, typically known as a 2P-Set (Two-Phase Set) or a PN-Counter (Positive-Negative Counter) for scenarios involving counters.
The 2P-Set is a state-based CRDT designed to handle both additions and deletions:
-
Two Sets: It comprises two sets:
- Add Set (A-Set): Stores all elements that have been added.
- Remove Set (R-Set): Stores all elements that have been removed.
- Add Operation: When an element is added, it is inserted into the A-Set.
- Remove Operation: When an element is removed, it is added to the R-Set. For an element to be considered removed, it must be present in both the A-Set and the R-Set.
- Membership Test: To determine if an element is in the 2P-Set, check if it is in the A-Set and not in the R-Set.
- Merge Operation: Merging two 2P-Sets involves taking the union of both the A-Sets and the R-Sets of the two CRDT instances.
This approach ensures that once an element has been added and subsequently removed, it cannot be re-added, as its presence in the R-Set permanently excludes it from membership, reflecting the “removal” operation.
For counters that need to handle increments and decrements (additions and removals), a PN-Counter can be used:
-
Two Counters: Consists of two separate counters:
- P-Counter: For positive increments.
- N-Counter: For negative decrements.
- Increment Operation: Increments are applied to the P-Counter.
- Decrement Operation: Decrements are applied to the N-Counter.
- Value Calculation: The actual value of the PN-Counter is computed as the difference between the P-Counter and the N-Counter.
- Merge Operation: Merging involves taking the element-wise maximum of both P-Counters and N-Counters from two PN-Counter instances.
The basic 2P-Set does not allow the re-adding of elements once they are removed because the R-Set permanently records their removal. To overcome this limitation and handle more complex scenarios (like re-adding previously removed items), a more advanced type of CRDT called an OR-Set (Observed-Removed Set) can be used:
- OR-Set: Each element in the OR-Set is associated with a unique tag when added, and instead of a single R-Set, a collection of tags is maintained for removed elements. This setup allows elements to be re-added with a new tag, distinguishing different instances of the same element over time.
These enhancements to the basic state-based CRDT models ensure that they can effectively handle the dynamics of add and remove operations in distributed systems, providing flexibility and robustness needed for real-world applications.
What happens to WAL when transaction is partially done with some writes made and the aborted due to subsequent writes failures?
In the scenario where a transaction needs to be aborted after a write has already been recorded in the Write-Ahead Log (WAL) but a subsequent write within the same transaction fails, the typical approach is not to remove the original entry from the WAL. Instead, the system handles this by adding an additional entry to indicate the abort operation. Here’s how this process generally works:
-
WAL Entries Are Immutable:
- Once a write operation is recorded in the WAL, it is considered immutable. This is crucial for ensuring the reliability of the log, especially for recovery purposes. Removing entries from the WAL could lead to inconsistencies and complicate the recovery process.
-
Logging the Abort:
- If a transaction needs to be aborted, the database system writes an “abort” entry into the WAL. This entry indicates that the transaction has been rolled back.
- During recovery, or when processing the WAL entries during normal operations, the presence of an abort entry means that the database should disregard the effects of any writes logged by the aborted transaction. If the changes were applied to the database before the abort (due to write-ahead logging), the system will use the WAL to undo those changes to ensure the database reflects a consistent state.
-
Undo Operations:
- Most database systems maintain enough information in the WAL to undo changes made by aborted transactions. This typically involves reversing the changes made by the transaction using the before-images or undo records stored in the WAL.
-
Transaction Rollback:
- When the abort entry is processed, the system performs a rollback operation. This includes undoing any changes that were applied to the database but not yet made permanent. The rollback operation ensures that the database remains consistent, reflecting only the effects of successfully committed transactions.
Consider a transaction that involves multiple write operations. Suppose the first write succeeds and is logged in the WAL, but the second write fails due to a constraint violation or some other issue:
- The system will log an abort entry into the WAL.
- Any changes made by the first write operation will be undone using the information from the WAL, maintaining database consistency.
- The transaction’s locks are released, and any resources held by the transaction are cleaned up.
- Reliability and Simplicity: Keeping the WAL entries immutable simplifies the design of the recovery process. The WAL needs to be a complete and accurate record of all attempts to modify the database, whether successful or not.
- Consistency: By writing abort entries and supporting comprehensive undo operations, the system ensures that it can always return to a consistent state, even after crashes or errors.
In database systems using WAL, the standard practice for handling transaction aborts after some operations have been logged is to add an abort entry to the WAL rather than attempting to remove or alter existing entries. This approach maintains the integrity and consistency of the WAL and by extension, the database itself, ensuring that it can recover correctly from any point of failure.
Real life examples of linearizability
Linearizability is a strong form of consistency that ensures all operations on a system appear instantaneously and atomically, preserving the order in which they occur. This model is critical for systems where operations must reflect a single, coherent order across all nodes, ensuring that any read reflects the most recent write. Systems requiring linearizability are often those where data accuracy and consistency are paramount for the correct operation and where users or processes might access the data concurrently from multiple locations.
-
Financial Systems:
- Banks and Trading Platforms: These require linearizability to accurately process transactions such as transfers, payments, and trades without discrepancies. For example, when transferring money between accounts, it is crucial that the debit from one account and the credit to another appear instantaneously to ensure that the accounts always reflect a correct and legal state.
- ATM Transactions: Ensuring that a balance check immediately after a withdrawal reflects the withdrawn amount.
-
E-commerce Platforms:
- Inventory Management: Ensuring that the inventory count is accurate when items are purchased online. Linearizability prevents the system from selling more items than are in stock, which is crucial during high-traffic periods like sales or promotions.
-
Distributed Databases and Data Stores:
- Systems like Google Spanner, which provide globally-distributed database services, rely on linearizability to ensure that all nodes show the same data at any time. This is crucial for applications that require global reach with consistent data views.
- NoSQL databases like Apache Cassandra can be configured for linearizable consistency (via Lightweight transactions) to ensure that certain operations, like conditional writes, are safely coordinated.
-
Real-Time Multiplayer Gaming:
- Ensuring that all players in a game see the same game state simultaneously. Linearizability is crucial for actions that have immediate consequences on the game state, influencing all players’ views and interactions.
-
Online Ticketing Systems:
- Systems used for booking seats at events or on flights must reflect changes universally the moment they occur. This prevents the system from double-booking a given seat.
-
Distributed Systems Providing Locks and Leader Election:
- Systems like Apache ZooKeeper use linearizability to manage distributed locks and leader election processes. These mechanisms ensure that, at any given time, all nodes in the system agree on which node holds the lock or who is the leader.
Benefits:
- Consistency and Simplicity: Linearizability simplifies the development and reasoning of distributed applications by providing a model that closely mimics a single, atomic system.
- Correctness: Guarantees the correctness of operations, especially in environments where decisions based on data state are critical.
Challenges:
- Performance and Scalability: Achieving linearizability can come at the cost of increased latency and decreased throughput, particularly in systems distributed across wide geographical areas.
- Complexity: Maintaining linearizability requires complex coordination mechanisms, often involving consensus protocols like Raft or Paxos, which can be challenging to implement and maintain.
Linearizability is crucial in environments where the accuracy and immediate visibility of operations are necessary for system integrity and user experience. While it provides strong consistency guarantees, the trade-offs in terms of performance and system complexity must be carefully considered and balanced against the specific requirements of the application.
How does Raft backfill stale nodes
When a leader is elected in Raft, it immediately takes responsibility for managing the log replication process across all nodes (followers). The leader maintains a nextIndex for each follower, which tracks the index of the next log entry the leader will send to that follower.
- When the leader attempts to replicate new log entries, it sends
AppendEntriesRPCs that include log entries starting from nextIndex for each follower. - Each
AppendEntriesRPC includes the term and the index of the entry immediately preceding the new entries being sent. This helps the follower to check whether its log is consistent with the leader’s.
- If a follower’s log does not contain the entry at the index and term specified by the leader (indicating a mismatch), the follower will reject the
AppendEntriesRPC. - Upon receiving a rejection, the leader decrements the nextIndex for that follower and retries sending the
AppendEntriesRPC, this time including earlier entries.
- This process continues, with the leader decrementing the nextIndex and resending
AppendEntriesRPCs, until the follower’s log entries match the leader’s log up to the point of divergence. - Once the logs are consistent (i.e., the follower has all the entries up to the current entry that the leader attempted to replicate), the leader then sends all subsequent log entries to the follower.
- The follower processes and appends each new entry to its log as it receives successful
AppendEntriesRPCs from the leader. - This method of decrementing the nextIndex and retrying ensures that even significantly stale nodes can be brought fully up-to-date by progressively backfilling all missing entries.
- It’s crucial that during this catch-up process, log entries are not only replicated but also committed. An entry is considered committed when it is safely replicated on a majority of nodes.
- The leader keeps track of which entries have been successfully replicated on each node, and updates the commitIndex accordingly. The commitIndex is the highest log index known to be committed.
- To optimize the process of catching up stale nodes, Raft implementations can include optimizations such as sending snapshots of the entire state up to a certain point if the node is extremely far behind. This is particularly useful when the missing entries are so numerous that sending them individually would not be sufficient
Relational database versus non-relational database
Many to many;
Joins; cross nodes/partitions read and write
Data locality; data denormalization.
Traditionally most no sql databases are non-relational however that’s not always the case.
MySQL versus PostgresSQL on isolation
The major differences between PostgreSQL and MySQL in terms of isolation levels primarily revolve around their implementation of isolation mechanisms, default behaviors, and supported features. Here’s a comparison:
-
Default Isolation Level:
- PostgreSQL uses Read Committed as its default isolation level.
-
Supported Isolation Levels:
- Read Uncommitted: PostgreSQL does not directly support this isolation level.
- Read Committed: Ensures that any data read during a transaction is committed at the moment it is read. This is the default.
- Repeatable Read: Ensures that if a transaction reads a row, it will see the same data if it reads that row again within the same transaction, preventing non-repeatable reads.
- Serializable: Provides full serializability using a technique called Serializable Snapshot Isolation (SSI). This level prevents all read and write anomalies, ensuring transactions are serializable without using traditional two-phase locking.
-
Concurrency Control:
- PostgreSQL primarily uses Multiversion Concurrency Control (MVCC), which allows readers to access a snapshot of the database at a point in time without being blocked by writers. Each transaction sees a consistent snapshot of the database, providing high concurrency and performance.
-
Serializable Snapshot Isolation (SSI):
- An advanced form of serializable isolation that detects conflicts and ensures serializable transactions without requiring extensive locking, reducing the likelihood of deadlocks and enhancing performance.
-
Default Isolation Level:
- MySQL uses Repeatable Read as its default isolation level.
-
Supported Isolation Levels:
- Read Uncommitted: Allows transactions to read uncommitted changes made by other transactions, which can lead to dirty reads.
- Read Committed: Ensures that any data read during a transaction is committed at the moment it is read, preventing dirty reads.
- Repeatable Read: Ensures that if a transaction reads a row, it will see the same data if it reads that row again within the same transaction, preventing non-repeatable reads. This is the default.
- Serializable: The highest isolation level, ensuring full serializability by locking the entire range of data being accessed, thus preventing all read and write anomalies. This can lead to higher contention and reduced concurrency.
-
Concurrency Control:
- MySQL also uses MVCC, particularly in the InnoDB storage engine, to provide non-blocking reads and support high concurrency.
-
Differences in Implementation:
- MySQL’s implementation of Repeatable Read differs from PostgreSQL’s as it maintains consistent snapshots of the database for the duration of a transaction, ensuring repeatable reads without phantom reads.
- MySQL’s Serializable isolation level is more traditional, using extensive locking to ensure serializable transactions, which can lead to higher contention and more frequent deadlocks compared to PostgreSQL’s SSI.
-
Default Isolation Level:
- PostgreSQL: Read Committed.
- MySQL: Repeatable Read.
-
Serializable Isolation:
- PostgreSQL uses Serializable Snapshot Isolation (SSI) to provide serializable transactions without heavy locking.
- MySQL uses traditional locking mechanisms, which can result in higher contention and deadlocks.
-
Support for Read Uncommitted:
- PostgreSQL does not support Read Uncommitted directly.
- MySQL supports Read Uncommitted, which allows for higher concurrency at the cost of potential dirty reads.
-
MVCC Implementation:
- Both databases use MVCC for high concurrency, but PostgreSQL’s MVCC implementation is known for providing better performance and consistency in complex transactional workloads.
By understanding these differences, you can make an informed decision on which database system better suits your specific application requirements and concurrency needs.
New types of Relational DB - Volt DB
Initially released in 2010.
- In memory hash index by default - not durable; WAL is optional; tree index is optional. Less data per node means more partitions and more cross partition reads and writes (distributed consensus for reads and writes - 2 phase commit).
- Actual serial execution with only a single thread. Bottle neck is the CPU.
- Minimize the network latency with stored procedures.
——————
In-memory relational database designed for real-time analytics and high-velocity transactional workloads.
It combines in-memory storage with ACID compliance, providing fast data processing and strong consistency.
VoltDB supports distributed architecture, enabling horizontal scaling and fault tolerance. It is particularly suited for applications requiring low-latency operations, such as financial services, telecommunications, and IoT.
Google spanner
Optimize read-only traffic by not using two phase commit but still satisfy linearizability.
Linerizability relies on real time not logical order! Therefore we cannot use lamport clocks!
Uncertainty period. Ensure no overlap.
GPS clock.
HDFS
Name node - metadata store.
* In-memory info about replicas store a certain file and files.
* WAL for recovery;
* snapshot the location/index of the WAL.
When the name node starts up, ask each data node which files it contains an and replicate file if necessary. Configurable number of replicas in each data node.
Read: ask name node for file location; name node respond the best replica - lower latency.
Write: picking the location of replicas, the name node tries to be rack aware and pick a primary node. Also name node is smart enough to do replication.
Map reduce
Early 2000; built to be super resilient to failures.
DDIA 399
MapReduce is a programming model and processing technique that allows for the scalable and efficient processing of large data sets across a distributed cluster of computers. Here are the key advantages of using MapReduce:
-
Scalability:
- Horizontal Scaling: MapReduce can easily scale to handle petabytes of data by distributing tasks across a large number of commodity machines. As data volumes grow, additional nodes can be added to the cluster to maintain performance.
-
Fault Tolerance:
- Automatic Recovery: MapReduce provides robust fault tolerance. If a node fails during processing, the system automatically reassigns the failed tasks to other nodes, ensuring that the job completes successfully.
- Data Replication: Input data is typically stored in a distributed file system like HDFS, which replicates data blocks across multiple nodes, further enhancing fault tolerance.
-
Simplified Programming Model:
- Abstraction: MapReduce abstracts the complexity of distributed computing, allowing developers to focus on the core business logic through the map and reduce functions without worrying about the underlying infrastructure.
- Ease of Use: The model simplifies the implementation of parallel processing tasks, making it easier for developers to write distributed applications.
-
Parallel Processing:
- Concurrent Execution: MapReduce processes data in parallel by dividing it into smaller chunks and distributing these chunks across multiple nodes. This leads to significant improvements in processing speed and efficiency.
-
Data Locality:
- Optimized Data Access: MapReduce moves computation to the data, meaning tasks are executed on the nodes where the data resides. This reduces network I/O and improves performance by leveraging data locality.
-
Flexibility and Versatility:
- Broad Use Cases: MapReduce can be used for a wide variety of data processing tasks, including ETL (Extract, Transform, Load) processes, data mining, machine learning, and log analysis.
- Language Support: Implementations of MapReduce (like Hadoop) support multiple programming languages, including Java, Python, and others, making it accessible to a wide range of developers.
-
Load Balancing:
- Efficient Resource Utilization: The MapReduce framework distributes tasks across the cluster in a balanced manner, ensuring optimal utilization of resources and preventing bottlenecks.
-
Integration with Big Data Ecosystem:
- Hadoop Ecosystem: MapReduce integrates seamlessly with other components of the Hadoop ecosystem, such as HDFS for storage, YARN for resource management, and tools like Hive and Pig for data querying and scripting.
Consider a scenario where a company needs to analyze log files generated by web servers to identify popular pages and user behavior patterns.
-
Map Phase:
- Each log file is split into lines.
- The map function processes each line, extracting relevant information such as the page URL and user ID.
- The map function emits key-value pairs, with the page URL as the key and a count of 1 as the value.
-
Shuffle and Sort:
- The framework automatically shuffles and sorts the key-value pairs based on the keys (page URLs).
- This phase groups all counts for the same URL together.
-
Reduce Phase:
- The reduce function aggregates the counts for each URL.
- It sums the values associated with each key, producing the total count of accesses for each page.
-
Output:
- The final output is a list of URLs with their corresponding access counts, stored in HDFS or another storage system for further analysis.
MapReduce offers several advantages for processing large data sets in a distributed and parallel manner:
- Scalability: Easily handles growing data volumes by adding more nodes.
- Fault Tolerance: Automatically recovers from node failures.
- Simplified Programming: Abstracts the complexities of distributed computing.
- Parallel Processing: Efficiently processes data in parallel.
- Data Locality: Reduces network I/O by processing data where it resides.
- Flexibility: Supports a wide range of data processing tasks.
- Load Balancing: Ensures efficient resource utilization.
- Integration: Works well with other tools in the big data ecosystem.
These advantages make MapReduce a powerful and versatile tool for big data processing and analytics.
Batch job data joins
DDIA 404
When we talk about joins in the context of batch processing, we mean resolving all occurrences of some association within a dataset. For example, we assume that a job is processing the data for all users simultaneously, not merely looking up the data for one particular user (which would be done far more efficiently with an index).
- Sort merge joins; can be extremely slow; have to sort all data by join key; have to send at least one whole dataset over the network.
- Broadcast hash joins; sends entire small dataset to all partitions and does a hash join. Don’t need to sort big dataset, small dataset must fit into memory;
- Partitioned hash joins; if both datasets are too big to fit into memory but they are partitioned the same way so that the partitions can fit into memory.
Why map reduce sucks
See in the picture
What are the common message broker service
Memory based
Disk based
Memory based message broker implementation
Use linked list.
Round robin delivery.
Log based message broker implementation
Consumer offset
Durability
Replay
Elasticsearch
Wrapper of Apache Lucene; APIs, query syntax, monitoring tools
Time series DB
Time scale DB; influx DB; Apache Drvid
column oriented table
As opposed to one large table with one big indexes, use many small indexes.
Hyper table.
Graph database
Neo 4J implementation
Geospatial indexes
Geo hashes / quad trees
Geo Sharding
Write back cache
Lowest latency writes
Data staleness/correct issues.
Reverse proxy versus load balancer
Load balancers and reverse proxies are both crucial components in modern web infrastructure, often used to distribute and manage network traffic. While they share some similarities, they serve different primary purposes and have distinct functionalities. Here’s a detailed comparison:
Purpose:
- A load balancer is designed to distribute incoming network traffic across multiple servers to ensure no single server becomes overwhelmed, enhancing the availability and reliability of applications.
Key Functions:
1. Traffic Distribution:
- Distributes client requests among multiple backend servers to balance the load and prevent any single server from becoming a bottleneck.
-
Improved Availability:
- Ensures high availability by redirecting traffic to healthy servers if one or more servers fail.
-
Scalability:
- Allows scaling out of resources by adding more servers to the backend pool, handling increased traffic seamlessly.
-
Health Monitoring:
- Continuously checks the health of backend servers to route traffic only to operational servers.
Types of Load Balancers:
1. Layer 4 Load Balancer:
- Operates at the transport layer (OSI Layer 4). It makes routing decisions based on IP address and TCP/UDP port.
- Examples: AWS Elastic Load Balancer (ELB), HAProxy (in Layer 4 mode).
-
Layer 7 Load Balancer:
- Operates at the application layer (OSI Layer 7). It makes routing decisions based on HTTP headers, cookies, URL paths, and other content data.
- Examples: Nginx, Apache HTTP Server, AWS Application Load Balancer (ALB).
Purpose:
- A reverse proxy is designed to act as an intermediary for requests from clients seeking resources from backend servers. It provides various functionalities such as load balancing, caching, SSL termination, and more.
Key Functions:
1. Load Balancing:
- Distributes incoming traffic among multiple backend servers (a common feature of many reverse proxies).
-
SSL Termination:
- Handles SSL/TLS encryption and decryption, offloading this resource-intensive process from backend servers.
-
Caching:
- Caches static and dynamic content to reduce the load on backend servers and improve response times.
-
Security:
- Acts as a security barrier, hiding the backend server details from clients and protecting against DDoS attacks, web threats, and other vulnerabilities.
-
Compression:
- Compresses outbound responses to reduce bandwidth usage and improve load times.
-
Content Routing:
- Routes requests based on URL paths, headers, and other request attributes.
Examples of Reverse Proxies:
- Nginx
- Apache HTTP Server
- HAProxy (in Layer 7 mode)
- Varnish
Load Balancer:
- Ideal for scenarios where the primary need is to distribute incoming requests across multiple servers to balance the load, ensure availability, and improve scalability. Common in high-traffic websites, enterprise applications, and cloud services.
Reverse Proxy:
- Suitable for scenarios requiring additional functionalities like SSL termination, caching, compression, security enhancements, and content-based routing. Often used in combination with load balancers for comprehensive traffic management and optimization.
While load balancers and reverse proxies can overlap in their functionalities, especially in traffic distribution, they serve different primary purposes. Load balancers focus on distributing traffic to ensure high availability and scalability, whereas reverse proxies provide a broader range of features including security, caching, SSL termination, and more. Depending on the specific needs of your infrastructure, you may use one or both in conjunction to optimize performance and manage traffic effectively.
Flow and congestion control
Help avoid overloading network buffer on receiver.
Web sockets
WebSockets is a communication protocol that provides full-duplex communication channels over a single TCP connection. Unlike traditional HTTP requests which follow a request-response model, WebSockets allow for persistent, two-way communication between the client and server. This makes WebSockets particularly suitable for applications that require real-time updates, such as live chat applications, online gaming, and real-time data feeds.
-
Full-Duplex Communication:
- WebSockets support full-duplex communication, allowing both the client and server to send and receive messages independently at any time.
-
Single TCP Connection:
- WebSockets use a single, long-lived TCP connection that remains open, enabling continuous data exchange without the overhead of establishing new connections.
-
Low Latency:
- By maintaining a persistent connection, WebSockets reduce the latency typically associated with establishing new connections for each message, resulting in faster communication.
-
Bidirectional Data Flow:
- Both the client and server can initiate communication, allowing for more interactive and responsive applications.
-
Handshake:
- The WebSocket connection begins with a handshake between the client and server. This handshake is initiated by the client using an HTTP request that includes an
Upgradeheader, requesting to switch the connection to the WebSocket protocol. - If the server supports WebSockets, it responds with an HTTP 101 status code and an
Upgradeheader, establishing the WebSocket connection.
- The WebSocket connection begins with a handshake between the client and server. This handshake is initiated by the client using an HTTP request that includes an
-
Data Frames:
- Once the connection is established, data can be exchanged in the form of frames. These frames are lightweight and can carry both text and binary data.
- WebSockets define a frame format that includes metadata such as the opcode (indicating the type of data) and payload length, allowing efficient data transfer.
-
Persistent Connection:
- The WebSocket connection remains open until either the client or server decides to close it. This persistent connection enables continuous, real-time communication.
-
Client Initiates Handshake:
- The client sends an HTTP request to the server with an
Upgradeheader:GET /chat HTTP/1.1 Host: server.example.com Upgrade: websocket Connection: Upgrade Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ== Sec-WebSocket-Version: 13
- The client sends an HTTP request to the server with an
-
Server Responds:
- The server responds with an HTTP 101 status code, upgrading the connection to WebSockets:
HTTP/1.1 101 Switching Protocols Upgrade: websocket Connection: Upgrade Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
- The server responds with an HTTP 101 status code, upgrading the connection to WebSockets:
-
Client Sends a Message:
- The client sends a message to the server using the WebSocket connection:
javascript const socket = new WebSocket('ws://server.example.com/chat'); socket.onopen = () => { socket.send('Hello, Server!'); };
- The client sends a message to the server using the WebSocket connection:
-
Server Receives and Responds:
- The server receives the message and can respond back through the same WebSocket connection:
javascript socket.onmessage = (event) => { console.log('Message from server:', event.data); };
- The server receives the message and can respond back through the same WebSocket connection:
-
Live Chat Applications:
- WebSockets enable real-time messaging between users in chat applications, providing a seamless and interactive user experience.
-
Online Gaming:
- In online multiplayer games, WebSockets facilitate real-time communication between players and the game server, ensuring low-latency interactions.
-
Real-Time Data Feeds:
- WebSockets are used to push real-time updates for stock prices, sports scores, and other continuously updating data streams.
-
Collaborative Tools:
- Tools like collaborative document editing and real-time drawing applications use WebSockets to synchronize changes between multiple users in real time.
-
IoT Devices:
- WebSockets provide a lightweight and efficient way for IoT devices to communicate with servers and other devices, supporting real-time data transfer and control.
- Low Latency: Maintains a persistent connection, reducing the overhead of establishing new connections.
- Efficient Communication: Supports full-duplex communication, enabling real-time data exchange.
- Reduced Server Load: Minimizes the number of HTTP requests, reducing server load and bandwidth usage.
- Scalability: Suitable for applications that require handling a large number of concurrent connections.
- Complexity: Requires more complex server and client implementations compared to traditional HTTP.
- Firewall and Proxy Issues: Some firewalls and proxies may not support or may interfere with WebSocket connections.
- Resource Management: Persistent connections can consume more server resources, requiring efficient management to handle large numbers of connections.
WebSockets provide a powerful and efficient protocol for real-time, bidirectional communication over a single TCP connection. They are well-suited for applications requiring low-latency interactions, such as live chat, online gaming, real-time data feeds, and collaborative tools. By maintaining a persistent connection, WebSockets reduce latency and improve the performance of interactive applications, making them an essential technology for modern web development.
Server sent events
Server-Sent Events (SSE) is a standard for server-to-client streaming of events. It allows servers to push real-time updates to web clients over a single HTTP connection. SSE is built on top of the HTTP protocol and is simpler to implement compared to WebSockets for certain use cases that require one-way communication from the server to the client.
-
One-Way Communication:
- SSE supports one-way communication from the server to the client. The server can push updates to the client, but the client cannot send data back to the server through the same connection.
-
Persistent Connection:
- The client establishes a persistent connection to the server, which remains open. The server can continuously send updates over this connection.
-
Text-Based Protocol:
- SSE uses a text-based protocol where events are sent as text messages. Each message can include fields such as
event,data,id, andretry.
- SSE uses a text-based protocol where events are sent as text messages. Each message can include fields such as
-
Automatic Reconnection:
- If the connection is lost, the client automatically attempts to reconnect to the server. The server can include a
retryfield to specify the reconnection delay.
- If the connection is lost, the client automatically attempts to reconnect to the server. The server can include a
-
Event Stream:
- Events are sent as a continuous stream of text data. Each event is separated by two newline characters, and each field within an event is separated by a newline character.
-
Client Initiates Connection:
- The client initiates the connection by sending an HTTP GET request to the server with the
Acceptheader set totext/event-stream. - Example:
http GET /events HTTP/1.1 Host: server.example.com Accept: text/event-stream
- The client initiates the connection by sending an HTTP GET request to the server with the
-
Server Responds:
- The server responds with an HTTP 200 status code and sets the
Content-Typeheader totext/event-stream. - Example:
http HTTP/1.1 200 OK Content-Type: text/event-stream
- The server responds with an HTTP 200 status code and sets the
-
Server Sends Events:
- The server sends events as text data over the open connection. Each event can contain multiple fields.
- Example:
```text
data: This is a messageevent: customEvent
data: This is a custom eventid: 12345
data: This is another message with an IDretry: 5000
data: This is a message with a custom retry interval
```
-
Client Handles Events:
- The client receives the events and processes them using JavaScript. The
EventSourceinterface is used to establish the connection and handle incoming events.
- The client receives the events and processes them using JavaScript. The
```javascript
// Create a new EventSource instance
const eventSource = new EventSource(‘/events’);
// Listen for messages
eventSource.onmessage = function(event) {
console.log(‘Message:’, event.data);
};
// Listen for custom events
eventSource.addEventListener(‘customEvent’, function(event) {
console.log(‘Custom event:’, event.data);
});
// Handle errors
eventSource.onerror = function(event) {
console.error(‘Error:’, event);
};
~~~
```javascript
const express = require(‘express’);
const app = express();
const port = 3000;
app.get(‘/events’, (req, res) => {
res.setHeader(‘Content-Type’, ‘text/event-stream’);
res.setHeader(‘Cache-Control’, ‘no-cache’);
res.setHeader(‘Connection’, ‘keep-alive’);
// Function to send an event
const sendEvent = (data, event = null, id = null, retry = null) => {
if (event) res.write(`event: ${event}\n`);
if (id) res.write(`id: ${id}\n`);
if (retry) res.write(`retry: ${retry}\n`);
res.write(`data: ${data}\n\n`);
};
// Send initial message
sendEvent('Connected to server');
// Send periodic updates
setInterval(() => {
sendEvent('Periodic update', 'update', Date.now().toString());
}, 10000);
// Handle client disconnect
req.on('close', () => {
console.log('Client disconnected');
res.end();
}); });
app.listen(port, () => {
console.log(Server running at http://localhost:${port}/);
});
~~~
-
Real-Time Notifications:
- SSE is suitable for applications that require real-time notifications, such as social media updates, chat messages, or system alerts.
-
Live Data Feeds:
- Applications that need to display live data, such as stock tickers, sports scores, or news feeds, can benefit from SSE.
-
Monitoring Dashboards:
- SSE can be used to push updates to monitoring dashboards, providing real-time metrics and status information.
-
Event Logs:
- SSE is useful for streaming event logs from servers to clients for real-time monitoring and analysis.
- Simplicity: SSE is simpler to implement compared to WebSockets, especially for one-way communication scenarios.
- Automatic Reconnection: The client automatically reconnects if the connection is lost, improving reliability.
- Text-Based Protocol: Easy to debug and understand, as events are sent as plain text.
- One-Way Communication: SSE only supports server-to-client communication. For bidirectional communication, WebSockets may be a better choice.
- Limited Browser Support: While widely supported, some older browsers may not support SSE.
- Message Size Limitations: SSE may have limitations on the size of individual messages, depending on the server and client implementations.
Server-Sent Events (SSE) is a protocol for server-to-client streaming of events over a single HTTP connection. It supports real-time updates, automatic reconnection, and a simple text-based format, making it ideal for applications that need continuous, one-way data flow from the server to the client. While it is easier to implement than WebSockets for certain use cases, it is limited to one-way communication and may have some compatibility and message size constraints.
Dockers and Kubernetes
Set the working directory
Overview:
Docker is a platform designed to make it easier to create, deploy, and run applications by using containers. Containers allow developers to package applications with all their dependencies and configurations, ensuring that the application runs consistently across different environments.
Key Features:
-
Containerization:
- Docker allows you to package an application and its dependencies into a single container. This container can run on any machine that has Docker installed, ensuring consistency and eliminating environment-related issues.
-
Isolation:
- Each Docker container runs in its own isolated environment. This isolation ensures that applications do not interfere with each other, enhancing security and stability.
-
Efficiency:
- Containers are lightweight and share the host system’s kernel, making them more efficient than traditional virtual machines (VMs). This allows for faster startup times and better resource utilization.
-
Portability:
- Docker containers can run on any system that supports Docker, whether it’s a developer’s laptop, a testing server, or a production environment. This portability simplifies the development and deployment process.
-
Version Control:
- Docker images (blueprints for containers) can be versioned, making it easy to track changes and roll back to previous versions if necessary.
Use Cases:
- Microservices architecture.
- Continuous Integration and Continuous Deployment (CI/CD).
- Development and testing environments.
- Running legacy applications.
Example:
A Dockerfile is used to define the environment and steps to build a Docker image.
```Dockerfile
# Use an official Python runtime as a parent image
FROM python:3.8-slim
WORKDIR /app
COPY . /app
RUN pip install –no-cache-dir -r requirements.txt
EXPOSE 80
ENV NAME World
CMD [“python”, “app.py”]
~~~
Overview:
Kubernetes (often abbreviated as K8s) is an open-source platform for automating the deployment, scaling, and management of containerized applications. It was originally developed by Google and is now maintained by the Cloud Native Computing Foundation (CNCF).
Key Features:
-
Orchestration:
- Kubernetes automates the deployment and management of containers, ensuring that the desired state of your application is maintained. It handles the distribution of containers across a cluster of machines, scaling them up or down as needed.
-
Scalability:
- Kubernetes can automatically scale your applications up or down based on demand, ensuring that your application can handle varying levels of traffic without manual intervention.
-
Self-Healing:
- Kubernetes continuously monitors the health of your application and can automatically replace failed containers, restart them if they crash, and reschedule them if a node in the cluster fails.
-
Load Balancing and Service Discovery:
- Kubernetes provides built-in load balancing to distribute traffic across multiple instances of your application. It also includes service discovery mechanisms to ensure that containers can find and communicate with each other.
-
Declarative Configuration:
- Kubernetes uses declarative configuration files (usually written in YAML or JSON) to define the desired state of your application. This makes it easier to manage complex deployments and version control your infrastructure.
Components:
- Pods: The smallest deployable units in Kubernetes, representing a single instance of a running process in a container.
- Services: Abstractions that define a logical set of pods and a policy to access them.
- Deployments: Controllers that manage the deployment and scaling of a set of pods.
- Nodes: Worker machines in a Kubernetes cluster, which can be physical or virtual.
Use Cases:
- Managing large-scale microservices architectures.
- Automating deployment and scaling of containerized applications.
- Ensuring high availability and fault tolerance.
- Simplifying DevOps workflows.
Example:
A simple Kubernetes deployment configuration for a web application.
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: webapp-deployment
spec:
replicas: 3
selector:
matchLabels:
app: webapp
template:
metadata:
labels:
app: webapp
spec:
containers:
- name: webapp
image: my-webapp-image:latest
ports:
- containerPort: 80
apiVersion: v1
kind: Service
metadata:
name: webapp-service
spec:
selector:
app: webapp
ports:
- protocol: TCP
port: 80
targetPort: 80
type: LoadBalancer
~~~
Docker:
- A platform for containerization, providing lightweight, isolated environments for running applications.
- Focuses on packaging applications and their dependencies into portable containers.
- Ensures consistency across different environments.
Kubernetes:
- A platform for orchestrating and managing containerized applications across a cluster of machines.
- Automates deployment, scaling, and operations of application containers.
- Ensures high availability, scalability, and fault tolerance.
Both Docker and Kubernetes are essential tools in modern DevOps practices, enabling the efficient development, deployment, and management of applications. Docker provides the foundation for containerization, while Kubernetes provides the tools to orchestrate and manage these containers at scale.
3 level of issues for isolation
Check the diagram
How is serializability different from linerizability
Serializability and linearizability are both consistency models used in distributed systems and databases to ensure correctness of concurrent operations, but they have distinct definitions and use cases. Here’s a detailed explanation of the differences between them:
Definition:
- Serializability is a property of a schedule (sequence of operations) where the outcome is the same as if the transactions were executed in some serial order, one after another, without any overlap in time.
- It is the highest level of isolation in database systems and ensures that concurrent transactions do not interfere with each other in a way that would produce incorrect results.
Use Case:
- Database transactions, where ensuring the correctness of concurrent operations is critical, such as financial transactions or inventory management.
Key Points:
1. Transactional Context:
- Serializability applies to a sequence of transactions, ensuring that the interleaved execution of transactions is equivalent to some serial execution of those transactions.
-
Anomalies Prevented:
- Prevents anomalies such as dirty reads, non-repeatable reads, and phantom reads.
-
Implementation:
- Can be implemented using two-phase locking, timestamp ordering, or other concurrency control mechanisms.
Example:
- Transactions T1 and T2:
- T1: Read(X), Write(X)
- T2: Read(X), Write(X)
- A schedule is serializable if the final state of the database is the same as if T1 and T2 were executed sequentially, either T1 followed by T2 or T2 followed by T1.
Definition:
- Linearizability (or atomic consistency) is a property of individual operations on a single object, ensuring that each operation appears to occur instantaneously at some point between its invocation and its response.
- It provides real-time guarantees about the order of operations.
Use Case:
- Distributed systems and concurrent data structures, where ensuring immediate consistency of operations on shared objects is critical, such as key-value stores, distributed queues, or counters.
Key Points:
1. Real-Time Constraints:
- Linearizability considers the actual time at which operations occur and ensures that the results of operations are consistent with real-time ordering.
-
Operation-Level Consistency:
- Applies to individual operations on single objects rather than entire transactions.
-
Anomalies Prevented:
- Prevents anomalies such as stale reads and ensures immediate visibility of writes.
Example:
- Operations on a shared register R:
- Operation O1: Write(R, 1)
- Operation O2: Read(R) -> 1
- A system is linearizable if O2 returns 1 only if O1 has been completed, reflecting the real-time ordering of operations.
- Serializability: Ensures a globally consistent order of transactions in a database. It is concerned with the correctness of interleaved transaction executions.
- Linearizability: Ensures immediate consistency of operations on a single object. It is concerned with the correctness of individual operations in real-time.
- Serializability: Ensures that transactions are executed in a way that is consistent with some serial order. It does not impose real-time constraints on the order of transactions.
- Linearizability: Ensures that operations reflect the real-time order in which they were applied. It provides a stronger guarantee than serializability by considering the exact timing of operations.
- Serializability: Typically implemented in databases to ensure the correctness of concurrent transactions. Common techniques include two-phase locking, timestamp ordering, and serialization graphs.
- Linearizability: Often implemented in distributed systems to ensure immediate consistency of operations on shared objects. Common techniques include consensus algorithms like Paxos and Raft.
Consider a distributed key-value store:
-
Serializability:
- Ensures that if two clients concurrently perform operations (e.g., one writes a value while another reads the value), the outcome is as if the operations were performed sequentially in some order.
- Does not guarantee that the read operation will immediately see the result of the write operation in real-time.
-
Linearizability:
- Ensures that if a client writes a value and another client reads the value, the read operation will see the most recent write immediately, reflecting the real-time order of operations.
- Provides a stronger guarantee that the value read is the latest value written.
- Serializability: Ensures that the outcome of executing concurrent transactions is equivalent to some serial execution of those transactions. It is used in databases to ensure transaction correctness without considering real-time constraints.
- Linearizability: Ensures that operations on a single object appear instantaneously and reflect real-time ordering. It is used in distributed systems to provide immediate consistency and real-time guarantees for operations on shared objects.
Understanding the differences between these consistency models helps in choosing the right approach for ensuring correctness and consistency in various applications, whether they are transaction-oriented databases or operation-oriented distributed systems.
How do we define event A happens before event B in distributed system?
Check picture
What’s the difference between physical clock and logical clock in distributed system?
Logical clocks are designed to capture casual relationships
What is gossip protocol
Check picture
What’s the way to achieve total order broadcast
Check pic
What’s idempotency
Check pic
When should we use which broadcast algorithm
Atomic commit versus distributed consensus
Check pic
Power of 2 to bytes
Check pic
Latency numbers that you should remember
Check out picture
Availability numbers
