Designing Data Intensive Applications (New Grad ↔ Mid Level)

At ZipRecruiter, I read Martin Kleppmann’s Designing Data-Intensive Applications over an entire year as a new grad software engineer aiming for mid level. These are the notes I kept as I worked through the book chapter by chapter.

Table of contents

• Part 1: Foundations of Data Systems
  • Chapter 1: Reliable, Scalable, and Maintainable Applications
  • Chapter 2: Data Models and Query Languages
  • Chapter 3: Storage and Retrieval
  • Chapter 4: Encoding and Evolution
• Part 2: Distributed Data
  • Chapter 5: Replication
  • Chapter 6: Partitioning
  • Chapter 7: Transactions
  • Chapter 8: The Trouble with Distributed Systems
  • Chapter 9: Consistency and Consensus
• Part 3: Derived Data
  • Chapter 10: Batch Processing
  • Chapter 11: Stream Processing

Chapter 1: Reliable, Scalable, and Maintainable Applications

• Applications are data intensive over cpu intensive

Thinking about Data Systems

• Databases and message queues both store data but have different access patterns
  • Group these into data systems regardless as multiple tools needed (memcached + elasticsearch) or one tool has multiple purposes (redis datastore/message queue)
• Data systems should be
  • Reliable: work correctly even when errors occur
  • Scalable: deal with system load growing
  • Maintainable: maintain current system behavior and be able to work on new features productively

Reliability

• TLDR works correctly even when any error occurs
• Systems that anticipate fault and deal with them → fault tolerant
  • Fault 1 aspect of system goes down and failure everything goes down
  • Good eng concept is to simulate faults so system deals with it correctly (Netflix Chaos Monkey)

Hardware Faults

• Hardware is reliable but accidents happen and disks die every 10-50 years
  • Can add redundancy with RAID, dual PSUs, backup generators but gets harder as you have larger amounts of machines → now deal with entire machine loss instead

Software Errors

• Software bugs occur and only prevented minimally with thinking about system interaction, testing, process isolation, chaos testing, measuring + monitoring behavior

Human Errors

• Human errors occur more than hardware errors and are unreliable
• Multiple ways to minimize this
  • Designing systems that minimize opportunities for error, creating sandbox environments to experiment with, tests from unit to integration, gradual code rollout, telemetry

How Important is Reliability?

• Very important and should be careful if cutting corners for development/cost cutting time

Scalability

• How to deal with increased system load/requests → making it scalable

Describing Load

• Load parameters amount per time (requests per second, cache hit rate)
• Example Twitter is 12k writes per second + distribution of followers per user
  • Can look up all people followed, find tweets for all followed people, and merge
  • Faster way maintain a home timeline cache and insert into users cache when follower posts
    • Good for reads but very expensive on tweet post with millions of followers → use approach 2 except when user has millions of followers

Describing Performance

• Throughput number of records processed per second and response time response - req time
  • Latency + processing time = response time
• Tail latencies are higher percentiles p95, p99, p999 and depending on use case are worth optimizing for → p99 major customers p999 factors out of control
• Service Level Agreements define expected performance and availability of service depending on median and p99, p999 percentile times
• Make sure to send requests regardless of response time to keep track of queueing delay/head of line blocking when testing load benchmark + keep rolling window of response times

Approaches for Coping with Load

• Vertical (increasing cpu power) vs horizontal (distributing load on multiple cpus) scaling
• Elastic systems add cpu resources on increased load
• Start from a single node and scale up only when increased load occurs and is needed

Maintainability

• Lots of software will be added onto for a while → make it easy for others to work on it

Operability: Making Life Easy for Operations

• Should have dedicated devops team and metrics for services

Simplicity: Managing Complexity

• More unattended complexity leads to snowballing effect of hard to understand software
• Abstract parts of software as needed to provide simplicity (SQL > disk operations and Programming Languages > Assembly, microservices)

Evolvability: Making Change Easy

• Agile working patterns give changeable framework providing tools and patterns for SWE

Chapter 2: Data Models and Query Languages

• Applications are built by layering 1 data model on top of another: features → data structures → bytes of memory on disk → current

Relational Model vs Document Model

• Relational Model is SQL where data stored in relations and relations are tuples
  • Did not have to worry about internal representation of data and most popular model

The Birth of NoSQL

• NoSQL invented for better scalability (write throughput/large datasets), open source, specialized query operations, dynamic data modeling
• Polyglot persistence: using both relational and non relational databases

The Object Relational Mismatch

• Programming is OOP but SQL is relational → impedance mismatch
  • ORMs help but do not alleviate this problem
• Json has better locality than sql as no need for performing multiple queries for nested information

Many to One and One to Many Relationships

• Using Id vs plain text allows you to standardize and deal with options along with not dealing with extra write overhead when fields are copied → database normalization
  • Easy for relational hard for non relational (multiple queries needed)

Are Document Databases Repeating History?

• Relationships 1 to N, N to 1, N to N
  • 1 to N: user has multiple properties
  • N to 1: users refer to a selected property connected by foreign key (ie selected city)
  • N to N: users refer to other users
• IBM information management system used hierarchical model (tree of nested records)
  • Worked for 1 to N relationships not N to N since no joins
• 2 solutions: network model and relational model came up
  • Network model every record now has N parents instead of 1 like hierarchical
    • Links stored as pointers and accessing record means following access path
    • Many paths lead to same record → need to keep track of all paths n dimensional search space
      • Querying and updating database complicated and inflexible + rewrite code needed for new access paths and hard to make data model changes
  • Relational model lays out all data as tuples identified via keys very openly
    • Query optimizer decides how to execute query and which indexes to use → declare index and software decides how to optimize

Relational Versus Document Databases Today

• Document databases used for schema flexibility, better locality, application data similar vs joins, 1 to N relationship and N to 1 and N to N relationships for relational databases
• Document models schema-less/schema on read where types enforce on read vs relational schema on write where all types are enforced everywhere in relational model
  • Runtime vs compile time checking in a sense
• Documents are good if need one time load and are small ; updates reupdate entire document

Query Languages for Data

• IMS/CODASYL is imperative and SQL is declarative
  • More concise, easier, hides implementation details of database engine, lends itself to parallelization since only specifies pattern of results

Declarative Queries on the Web

• CSS is declarative, DOM manipulation is imperative

MapReduce Querying

• MapReduce is based on the 2 functions map/collect and reduce/fold where loop over the data
  • Create key value pairs with map, group by key, and for every key run the reduce function
  • Functions must be pure

Graph Like Data Models

• If there are many to many like data models in data, makes sense to use graph data model even with different data types per node

Property Graphs

• Every vertex contains uid, outgoing edges, incoming edges, properties
• Every edge contains uid, vertex where edge starts and ends (tail/head vertex), label for relationship, properties

The Cypher Query Language

• Declarative query language for property graphs for Neo4j

// Query to insert graph data
// Create Nodes then (Node 1) -[:label1]-> (Node2) ...
CREATE
 (NAmerica:Location {name:'North America', type:'continent'}),
 (USA:Location {name:'United States', type:'country' }),
 (Idaho:Location {name:'Idaho', type:'state' }),
 (Lucy:Person {name:'Lucy' }),
 (Idaho) -[:WITHIN]-> (USA) -[:WITHIN]-> (NAmerica),
 (Lucy) -[:BORN_IN]-> (Idaho)

// Finding people who emigrated from USA to Europe
MATCH
 (person) -[:BORN_IN]-> () -[:WITHIN*0..]-> (us:Location {name:'United States'}),
 (person) -[:LIVES_IN]-> () -[:WITHIN*0..]-> (eu:Location {name:'Europe'})
RETURN person.name

• Multiple ways to express a query to get the same result

Graph Queries in SQL

• Can express graph queries in SQL with recursive common table expressions

WITH RECURSIVE
 -- in_usa is the set of vertex IDs of all locations within the United States
 in_usa(vertex_id) AS (
 SELECT vertex_id FROM vertices WHERE properties->>'name' = 'United States'
 UNION
 SELECT edges.tail_vertex FROM edges
 JOIN in_usa ON edges.head_vertex = in_usa.vertex_id
 WHERE edges.label = 'within'
 ),
 -- in_europe is the set of vertex IDs of all locations within Europe
 in_europe(vertex_id) AS (
 SELECT vertex_id FROM vertices WHERE properties->>'name' = 'Europe'
 UNION
 SELECT edges.tail_vertex FROM edges
 JOIN in_europe ON edges.head_vertex = in_europe.vertex_id
 WHERE edges.label = 'within'
 ),
 -- born_in_usa is the set of vertex IDs of all people born in the US
 born_in_usa(vertex_id) AS (
 SELECT edges.tail_vertex FROM edges
 JOIN in_usa ON edges.head_vertex = in_usa.vertex_id
 WHERE edges.label = 'born_in'
 ),

  -- lives_in_europe is the set of vertex IDs of all people living in Europe
 lives_in_europe(vertex_id) AS (
 SELECT edges.tail_vertex FROM edges
 JOIN in_europe ON edges.head_vertex = in_europe.vertex_id
 WHERE edges.label = 'lives_in'
 )
SELECT vertices.properties->>'name'
FROM vertices
-- join to find those people who were both born in the US *and* live in Europe
JOIN born_in_usa ON vertices.vertex_id = born_in_usa.vertex_id
JOIN lives_in_europe ON vertices.vertex_id = lives_in_europe.vertex_id;

Triple Stores and SPARQL

• 3 part state : (subject, predicate, object) : (I, eat, food), (Monkey, eats, banana)

  • subject is vertex and either predicate is edge + object is vertex OR key value property of {predicate, object}

  @prefix : <urn:example:>.
  _:lucy a :Person; :name "Lucy"; :bornIn _:idaho.
  _:idaho a :Location; :name "Idaho"; :type "state"; :within _:usa.
  _:usa a :Location; :name "United States"; :type "country"; :within _:america.
  _:namerica a :Location; :name "North America"; :type "continent".

• Triple store independent from semantic web which proposes the idea that if websites publish text info, why not publish machine readable data for computers (create web of data RDF ie Resource Description Framework)

  • Very complicated and multiple variations so no implementation

• RDF can be expressed in turtle language, xml, and more

• SPARQL query language for triple stores with RDF model similar to Cypher

SELECT ?personName WHERE {
 ?person :name ?personName.
 ?person :bornIn / :within* / :name "United States".
 ?person :livesIn / :within* / :name "Europe".
}

The Foundation: Datalog

• Datalog foundation as query language for Datomic and Cascalog —Datalog implementation for querying Hadoop datasets

-- First define triples of predicate(subject, object) NOT (subject, predicate, object)
name(namerica, 'North America').
type(namerica, continent).

name(usa, 'United States').
type(usa, country).
within(usa, namerica).

name(idaho, 'Idaho').
type(idaho, state).
within(idaho, usa).

name(lucy, 'Lucy').
born_in(lucy, idaho).

-- Then write query
within_recursive(Location, Name) :- name(Location, Name). /* Rule 1 */

within_recursive(Location, Name) :- within(Location, Via), /* Rule 2 */
																		within_recursive(Via, Name).

migrated(Name, BornIn, LivingIn) :- name(Person, Name), /* Rule 3 */
																		born_in(Person, BornLoc),
																		within_recursive(BornLoc, BornIn),
																		lives_in(Person, LivingLoc),
																		within_recursive(LivingLoc, LivingIn).
?- migrated(Who, 'United States', 'Europe').
/* Who = 'Lucy'. */

Chapter 3: Storage and Retrieval

• We should know what the database engine is doing under the hood in order to decide between them

Data Structures That Power Your Database

• Mock database with KV pairs in log file has good set performance but bad get O(n) performance
  • Solve this with indices which speed up read queries but slow down write queries as index needs to be updated

Hash Indexes

• For every key store the byte offset of the data in disk in a hash-map
  • Good for bulk writes on few keys to store all keys in memory
• Don’t run out of disk space by storing logs in segments and compacting them to only keep most recent key updates + merge segments after compaction in background
  • Faster and simpler to use binary over CSV for logs
  • To delete, add deletion record to data file and while merging, process discards any value with deletion record
  • Store snapshots of hash index on disk in case crash happens to load into memory quickly
  • Store checksums to detect corrupted logs and records
  • 1 writer thread for concurrency control and segments immutable aside from appending
    • Sequential writes faster than random writes, simpler concurrency/crash recovery, constant merging is why append > update in place
• Must fit hash index in memory + range queries from X to Y where X ≤ Y inefficient

SSTables and LSM-Trees

• Sorted String table mandates segments are sorted by key + every key appears once per segment

  • Easier merges (merge from mergesort) and easier searches as keys in hash index are sorted

  • Read requests traverse multiple grouped keys, put adjacent values in blocks and compress them before disk writing

    

• To make SSTables, maintain balanced BST, send R/W queries to it and once full, write to SSTable and check BST and SSTable for queries + background merge and compaction process

  • Will lose BST on DB crash so create append log of all writes as backup for memtable

• Used in levelDB, rocksDB, Cassandra, HBase but originally called Log Structured Merge Tree

• Bad performance when looking up keys that do not exist so pair with bloom filter

• Size tier compaction merges smaller SSTables into bigger ones and level tier compaction moves older data into levels top to bottom where top is most recent

B-Trees

• Default index implementation in SQL Databases

• Tree with sorted keys where every node has max BRANCHING_FACTOR references

  

• Page references are intact when overwriting pages and overwriting considered hardware operation so create write ahead log to prevent any crash mishaps

  • If accessing tree with multiple threads use latches (lightweight locks)

• Multiple optimizations can be made for BTrees

  • Instead of write ahead log, use copy on write for modified pages
  • Abbreviate keys to save space, add leaf next/prev pointers, try to make leaf pages appear in sequential order in disk

Comparing B-Trees and LSM-Trees

• LSM Trees faster for writes and BTrees faster for reads
• LSM Trees have higher write throughput as it has lower write amplification (one write to database resulting in multiple writes to disk) + sequential compaction + compressed better due to not being page oriented
  • SSDs have LSM algorithms to turn random writes into sequential but less write amplification and fragmentation allows more read/write throughput
• LSM trees though have to wait for compaction to finish before read/write requests happen so BTrees are more predictable here
  • Disk bandwidth shared between writes and compaction threads → if write throughput gets too high then compaction does not keep up leading to multiple unmerged segments → reads check more segment files and slow down
  • Keys may be in many segments vs 1 in B Tree so less strong transactions here

Other Indexing Structures

• Key/Value indexes are primary key indexes in relational model and secondary indexes can be created manually with CREATE_INDEX
  • Secondary keys aren’t unique so add matching row identifiers or append key row identifiers
• Heap files store row data in any order and can be overwritten in place given new data ≤ old data size
  • non clustered heap file structure: key 1 → pointer to heap file data → {row 1 : data 1}
  • clustered heap file structure to avoid index to heap file hop: key 1 → {row1 : data 1}
  • covered index heap file stores necessary info in index w/o accessing heap file: key 1 → {n1, n2, ..} + heap file
• Concatenated indexes makes tuples act like keys in sorted order
  • If we index (lastname, firstname) → phone number we can find people with the particular lastname, firstname or lastname but not firstname as it is not sorted
  • Helps query several columns at once
    • Can query col1, col2, col … up to n and any prefix of this and nothing else
• Can expand index by using fuzzy searches to search for similarly worded keys
• People tolerate slow disks as big and cheap but cheaper RAM makes in memory databases possible
  • Memcached used for caching but can add durability with battery powered RAM w/ logs, disk snapshots, replicating in memory state as needs to reload state on restart (Memsql)
  • No need to serialize data in memory compared to disk database + data models offered that disk indexes find hard to implement (pq and sets)
  • Can do anti caching approach if dataset larger than memory and evict LRU data to disk similarly how OS evicts and gets back virtual memory

Transaction Processing or Analytics

• Applications that are interactive and looks over a few records by key are online transaction processing (OLTP)
• Applications that scan over a lot of records but read a few columns and then calculate a statistic is online analytic processing (OLAP)
• SQL is good for both queries but can also run OLAP on separate database ie data warehouse

Data Warehousing

• OLTP databases are complex and autonomous and need to deliver immediate performance so don’t want to run OLAP queries on this as it could compromise performance + can optimize OLAP queries on a separate database

• Data warehouse is read only copy of OLTP databases which is extracted, transformed into analysis friendly schema, cleaned up, loaded in data warehouse Extract Transform Load (ETL)

  

• Used more in large companies as smaller companies have few OLTP systems and less data in each

• SQL services are diverging and are optimized towards OLTP or OLAP but not both

Stars and Snowflakes: Schemas for Analytics

• Star schema has 1 row per event and for every column store info about the product (foreign keys to dimension tables/numeric metrics)

  • Dimension table when where what who how why of the event

  

• Snowflake schema extension of star schema where dimensions are broken down into sub-dimensions → more normalized but harder to work with

Column-Oriented Storage

• Data warehouse queries only access a few columns at a time

  • Can have indexes on secondary keys but row oriented storage engine loads all rows from disk into memory → parse them + filter

  • Column oriented storage helps and store all column based values together so that query only parses column files

    

Column Compression

• Many column values are the same so use bitmap encoding in order to compress data

• For every column value, store a bitmap of length(columns) and 0 if value not at column[i] and 1 if value at column[i] if small or run length encode

  • Works very well for selecting individual column values ie WHERE x in (1, 2, 3)

  

• Other bottlenecks for scanning over millions of rows include efficient memory to cpu cache bandwidth, branch prediction avoidance, cpu instruction processing pipeline bubbles, using SIMD

  • Compressed data allows query engine to fit data in L1 cache and operate on data correctly ie vectorized processing

Sort Order in Column Storage

• To sort columns in column oriented storage we still sort rows and apply column storage as we need to reconstruct rows somehow with column keys as sort keys
  • Helps with data compression as run length encoding is much smaller with fewer distinct values
  • Good to keep the same data sorted in multiple ways to optimize for query patterns along with preventing data loss

Writing to Column-Oriented Storage

• Cannot update in place like BTrees as updating a row in sorted table would mean rewriting all column files
• Use LSM tree approach where write to sorted in memory store and periodically merge into column files

Aggregation: Data Cubes and Materialized Views

• If many aggregates are used such as MAX, COUNT, AVG. etc why not cache them → materialized aggregate

  • Materialized view is a table whose contents are the result of queries where engine reads from along with processing the rest of the query on the fly
    • Needs to be updated when data changes so bad for OLTP but good for OLAP

• Data/OLAP cube allows you to store all aggregates of a table depending on dimensions

  

Chapter 4: Encoding and Evolution

• When schema changes need to deploy new application code but not instant
  • Rolling upgrade/staged rollout deploys to few nodes at a time and spreads for server and client need to wait till user installs upgrade
• Compatibility needs to be maintained as many versions of code exist
  • Backwards compatibility new versions can work with old clients/data → easy
    • Server upgrades first, clients lag behind
  • Forwards compatibility old versions can work with new clients/data → hard as old code needs to ignore new changes
    • Client stays old, server moves ahead

Formats for Encoding Data

• We already store data in memory w/ pointers but we need to encode/decode it when sending it over the network as another process does not understand a local pointer

Language Specific Formats

• Language support for encoding/decoding in memory objects into bytes exist but they are restricted to same language services only
  • Decoding process needs to instantiate arbitrary class to restore data → attackers can decode arbitrary bytes → attackers can instantiate arbitrary classes → attackers can execute arbitrary code
  • No versioning available along with bad performance makes this a bad idea

JSON, XML, and Binary Variants

• Use standardized encodings readable like everyone like JSON/XML but still have various problems

  • Number ambiguity, large numbers not parsed properly, no binary string support, complex schemas, no schema for CSV so manual changes must be separately implemented

• Many binary encodings for JSON made to compress data

  

Thrift and Protocol Buffers

• Make schema then use code generation tool to generate classes to implement schema in code

• Thrift BinaryProtocol does not have field names but field tags to compress and Thrift CompactProtocol packs field type and tag number in 1 byte and protobufs are similar to Thrift CompactProtocol

  

  

  

• Thrift and Protobufs only rely on field tags to identify records so you cannot change field tags only field names and values

  • Can add new fields to schema with new tag numbers for forward compatibility
  • Can read older fields in newer schemas for backwards compatibility but cannot make required fields as could read old code without the field
  • Can only remove optional fields and cannot use same tag number again

• Can change certain types but can lose precision (32 bit to 64 bit integer)

Avro

• More compact schema as no tag numbers
  • Need to go through fields in order of schema and use schema to tell u datatype of fields as nothing to identify fields or datatype → code reading data must use exact same schema as code writing data

// JSON Equivalent
{
	 "type": "record",
	 "name": "Person",
	 "fields": [
		 {"name": "userName", "type": "string"},
		 {"name": "favoriteNumber", "type": ["null", "long"], "default": null},
		 {"name": "interests", "type": {"type": "array", "items": "string"}}
	 ]
}
// AVRO IDL
record Person {
 string userName;
 union { null, long } favoriteNumber = null;
 array<string> interests;
}

• Writer/encoder and Reader/decoder schema only need to be compatible not exact same and translates writer schema data into reader schema and ignores/fills with default fields as needed
• Can only add field with default value to maintain forwards/backwards compatibility as adding field with no default value means new readers cant read from old writers and removing field with no default value means old readers can’t read from new writers
• Reader knows writers schema by attaching it once at beginning of file, version numbers to schema mapping, negotiating schema version on connection setup
• Avro no tag names as its used for dynamically generated schemas compared to Protobufs/Thrift where one would have to assign field tags normally
  • Codegen is frowned upon dynamically typed languages since no compile time checker so Avro is favored here

The Merits of Schemas

• We use schemas over JSON/XML/CSV due to better compactness, forms of documentation, forward/backwards compatibility check, code generation and type checking at compile time

Modes of Dataflow

• Data flows everywhere from process to process or through databases, service calls, async messages

Dataflow Through Databases

• Process encodes then writes to database and process decodes when reading from it
• Backwards compatibility is implied along with forwards as old version of code could read from newer database as multiple processes can access database
  • Must be careful to not lose newer fields in older code translations on app level
• Data always outlives code and newer fields in older rows are defaulted to null on read and database snapshots for warehouse is encoded with latest schema

Dataflow Through Services: Rest and RPC

• Microservices are larger applications decomposed into smaller services
• Webservices include client HTTP requests, inter-org communication, making requests to external org
• REST builds on HTTP using urls for identification and uses this for cache control/auth/etc
• SOAP XML based independent from HTTP with API descriptions via Web Services Description Language ie WSDL
• RPCs across networks have problems compared to local function calls which includes network specific errors, no response received, responses/requests lost, higher latency, serialization overhead, language to language translation for datatypes but are still used
  • More performant than JSON over REST but harder top debug, less compatibility, less ecosystem available compared to REST
• Servers get updated first then clients so backwards compatibility on requests and forwards compatibility on responses

Message Passing Dataflow

• Async message systems in the middle of RPC and databases as they deliver to another process fast and also sent via an intermediary called message queue
  • Acts as buffer to prevent overloading, automatically redelivers messages as needed, sender does not need to know destination ip/port + decoupling, send to multiple at once vs RPC
  • Message passing is 1 way, to get a response you would read from another message queue
• Process ends message to queue and broker delivers to consumer
• Actor models instead of using threads put logic in actor process and process messages from actor to actor ie node to node which is scalable across multiple nodes
  • Distributed actor framework is message queue + actor model
• An algorithm is correct when it satisfies all properties defined in the system model
  • Safety properties a property has been broken at some time that can be pinpointed and must always hold
  • Liveness properties may not hold currently but it may be satisfied in the future
  • However algorithms assume some cases that will not hold in theory but we still need to account for them in practice
  • Safety and liveness are independent properties but can make tradeoffs around them
    • Can sacrifice safety for liveness meaning that we can let the system continue but it may end up in a bad state
    • Can sacrifice liveness for safety meaning that we don’t let the system continue but it is safe

Chapter 5: Replication

• Replication used for closeness/latency reasons, increases availability, scales out number of machines for queries
  • Easy if data does not change but harder to deal with if it does along with does data fit in dataset or not (sharding or no sharding)

Leaders and Followers

• Need to ensure replicas get all the data so use master slave replication where masters get writes to the database → they write to local storage → slaves receive replication logs
  • Masters can only be read/write and slaves get only reads

Synchronous Versus Asynchronous Replication

• Asynchronous replication is preferred as if 1 replica is down/recovering/slow then leader blocks all writes until resolved slowing down the entire system
  • Asynchronous replication on databases means only 1 follower is synchronous and rest are asynchronous ie semi synchronous

Setting Up New Followers

• To setup new followers cannot copy files from 1 node to another as both can be in different states and locking database reduces availability
• Instead take snapshot of leader database, copy that to follower, and follower requests leader of all changes since its latest log sequence numbers and processes

Handling Node Outages

• If follower goes down then request leader give changes from last log number
• If leader goes down then a follower must be leader then clients need to reconfigure and send writes to new leader and followers consume data changes from new leader → failover
  • Determine leader has failed due to polling messages, choose leader through election process, make system use new leader
  • Conflicting writes in async replication between new and old leader should be discarded but weakens durability
• Timeout value is important as too long timeout and more time to recover but too short timeout and could be false positives

Implementation of Replication Logs

• Can do statement based replication where leader sends over every SQL query it executes to its followers however nondeterministic values like RAND() or UPDATE prev data make it impossible to use unless you precompute and send the values over
• Can also write to a log first to leader and send it over to followers describing which bytes to change in which disk blocks
  • Replication is closely related to storage engine so if storage format changes then can’t run different versions of database
    • Solve this by upgrading the followers versions then reassigning one of the followers as the leader
• Can use log based replication which decouples log from storage engine and records at a high level allowing backwards compatibility and easier to parse
  • For insertions log the new values of all columns, deletions log primary key or columns if no primary key, updates log primary key and new values of columns
• Application code developed replication needed for specific cases over database system replication
  • Can use triggers to register application code that executes when transactions happen in database which logs changes in separate table albeit lower performance and buggier

Problems With Replication Lag

• Replication good for node failures, scalability, and latency and writes must go through single leader and reads any replica
• If few writes and many reads, create many followers and asynchronously replicate ; synchronous replication would make system unavailable to write if 1 failure
  • However writes take time so reading from follower could lead to outdated information with replication lag
  • If no writes done then followers eventually catch up to leader → eventual consistency

Reading Your Own Writes

• Imagine writing some data and viewing it right after → with async replication the new data hasn’t hit replica so looks like data is lost
  • Must also consider replicas being within data centers so must find datacenter then replica/leader
  • Solution is read after write consistency → only make sure that the user can see their updates on reload but no guarantees for other users views
• To implement this we can:
  • Read from leader if its something user has modified
  • For a set time after an update, read from the leader if many things are editable by user or prevent queries on followers after a set lag delay amount
  • Can remember timestamp of most recent write client side and make sure replica getting reads has been updated to at least that timestamp
• If we have multiple devices we also want to make sure users can see updates on multiple devices right after → cross device read after write consistency
  • Need to centralize metadata relating to timestamps + route same device requests to same datacenter

Monotonic Reads

• Imagine making multiple read requests in a row to replicas that are being updated → one replica could have fully updated information and others could have incomplete information due to replication lag
  • Can enforce monotonic reads in which if users make multiple read requests then they will not read older data after reading newer data → enforce this by making sure every user reads from the same replica always

Consistent Prefix Reads

• If different replicas/followers lag by different amounts this may break casualty where we receive inferred responses based on laggy sources
  • Consistent prefix reads state that if writes happen in a certain order then anyone seeing those rights see them appear in the same order

Solutions for Replication Lag

• Can trust application to read from leader but this is application dependent, why not trust db overall and provide transactions for stronger guarantees
  • Single node transactions but expensive for distributed databases and eventual consistency will always arise there

Multi-Leader Replication

• Need to allow multi leader replication as single point of failure for 1 leader so call this master/master replication

Use Cases for Multi-Leader Replication

• Useful for spreading out between multiple data centers where every datacenter has a leader
  • Better latency as writes go to local datacenter first and asynchronously replicated to other data centers vs network latency finding the only datacenter to write to
  • Every datacenter can operate independently of each other so no need to do failover operation
  • Internet is less reliable than local datacenter networks so multi leader replication deals with network problems better
• However data may be concurrently modified in 2 different data centers so wandering in dangerous territory
• Another use case of multi leader replication is for devices with apps disconnected from the internet (calendar, email)
  • Need to be able to view, modify, and sync with server and receive updates with the devices local database acting as a leader
• Real time collaborative editing apps has local replica asynchronously replicated to server and other users replicas editing the same document
  • No editing conflicts means lock on document needs to occur ⇒ single leader replication with transactions or record single keystrokes and avoid locking for multi leader replication

Handling Write Conflicts

• Write conflicts can happen in multi leader replication imagine if value set from A→ B on 1 leader and A→C on another and this conflict gets detected later
  • Does not happen on single leader replication as second writer blocks
• One solution is to avoid conflicts by making writes of a record go through same leader but if need to change leader for a record then have to deal with concurrent writes again
• Can instead converge towards a consistent state a couple of ways
  • Only take writes with highest timestamp (last write wins) or let higher number replica writes precede but leads to data loss
  • Can merge both conflicting values together or record conflicts in data structure and resolve later
• Can resolve conflicts later on write by calling custom handler or on read receive multiple versions of data row by row/document by document

Multi-Leader Replication Topologies

• Replication topologies describe communication paths on which writes travel
  • Can have circular, star topologies but most general is all to all where nodes send writes to all other nodes with unique node ids given to prevent replication loops
  • Circular/star topologies if 1 node fails then entire replication flow is interrupted and requires manual intervention which is why all to all is preferred
    • However some network links in all to all are fast and may overtake others so use version vectors to solve this

Leaderless Replication

• Leaderless replication which Dynamo uses allows any replica to accept writes from clients where client sends writes to several replicas or a coordinator node does this without enforcing write ordering

Writing to the Database When a Node is Down

• Failover does not exist in leaderless replication, instead send write to all replicas, enforce a replica count acknowledged policy, get all replica responses and take latest response
• To get all data when replica comes back online we:
  • Read repair where we send back replicas sending stale reads the correct value
  • Anti entropy process looks for replica data differences and copies data from 1 replica to another in no specific order
• If there are n replicas, w nodes must confirm a write, and r nodes confirm a read, then as long as w + r > n, we have quorum reads and writes

  • w = r = (n + 1)/2 where n is odd

    

Limitations of Quorum Consistency

• Smaller w and r values more likely to read stale values but you get lower latency and higher availability
• Sloppy quorums and concurrent writes limit quorum consistency unless you want to merge concurrent writes as clock skew can lose writes if using last write wins
• If concurrent write happens with a read, don’t know if read most recent value
• Nodes can succeed on replicas and fail on others, data restoration with a node with new value fails w condition, and overall weaker guarantees here
• Can monitor replication lag normally but in leaderless replication writes appear in no order so cant use so hard to track

Sloppy Quorums and Hinted Handoff

• Sloppy quorums (the latter) state is it better to return errors for all non quorum requests or should we accept writes anyways and write to all available nodes
  • Nodes are still w and r but not the originally selected nodes so these temporary nodes send to the correct nodes once available (hinted handoff)
  • Increases write availability but now you don’t know if reading latest value
• Can add multi datacenter support to leaderless model by specifying how many replicas per data center and first propagating from local to abroad data centers

Detecting Concurrent Writes

• Can have conflicting writes in leaderless databases so must deal with it
  • Last right wins does converge correctly but weakens durability as it loses n - 1 replicas data for n concurrent writes and must only write key once and not concurrently if you want to be safe
• Important to recognize that concurrent operations happen at the same time vs causal operations which one operation depends on the other
• Utilize version numbers per key and on read the client gets all values not overwritten and on write must include version number and merge values received in a previous read and only overwrite values with a lower version number
  • Merge siblings and if something will be deleted mark it with a tombstone to let clients know

### strongest → weakest consistency
Linearizable
  ⬇
Sequential
  ⬇
Causal consistency
  ⬇
{ Consistent prefix
  Monotonic reads
  Read your writes }
  ⬇
Eventual consistency

Chapter 6: Partitioning

• Datasets get so large that we may need to break them up into partitions for scalability
  • Can also parallelize execution across multiple partitions

Partitioning and Replication

• Copies of partition stored on multiple nodes for fault tolerance with nodes having multiple partitions

  

Partitioning of Key-Value Data

• Want to avoid skewed partitions resulting in hot spots so need to tune the partition algorithm
  • Don’t partition randomly as no way to know which partition to access when reading

Partitioning by Key Range

• Can partition by sorted key ranges similar to dictionary a-c, d-f, etc and can automatically rebalance boundaries to maintain evenness
  • However can still have constantly accessed hotspots while others collect dust so need to carefully select

Partitioning by Hash of Key

• To avoid skew can also partition by hash of key and partitions sorted by sorted hash ranges but now lose sorted range queries
  • Compromise with cassandra with compound primary key (containing columns) where first part of key/column is hashed and rest of column data sorted
• Partitioned evenly shaped or chosen randomly (consistent hashing)

Skewed Workloads and Relieving Hot Spots

• No matter what we still end up with hot spots e.g. a celebrities tweet being liked multiple times so either take that load or split up the data into multiple keys and make reads read all keys and combine result so need to only do this for few hot keys

Partitioning and Secondary Indexes

• How to deal with secondary indexes for partitions since they don’t map neatly to partitions?

Partitioning Secondary Indexes By Document

• Can partition by document only but in order to get all secondary index data, must read from all partitions and gather data (scatter gather pattern) which is expensive

Partitioning Secondary Indexes by Term

• Just like primary indexes, we can partition on the term value either by range or by hash which makes reads more efficient since no scatter gather but writes slower as multiple partitions of index affected on write
  • Updates to multiple partitions would be distributed and asynchronous so take some time to update or db might not support this

Rebalancing Partitions

• As hardware changes, you need to move around data/requests and rebalance nodes and partitions
  • Expected for database to continue accepting requests while rebalancing, minimize rebalancing, and load should be shared normally after rebalancing

Strategies for Rebalancing

• Don’t use hash % N for partitioning since if number of nodes change then more keys will have to be rebalanced as hash % N changes most of the keys

• Instead create more partitions than nodes and new nodes take a few partitions from other nodes until fairly distributed

  

• Number of partitions should be fixed but if you choose too many partitions(smaller partitions) then high overhead and if dataset size is variable and keeps growing, large partitions take more time to recover

• Key range partition are inconvenient for fixed number of partitions since cooked if all data in 1 boundary → dynamically increase or decrease number of partitions and adjust around nodes if necessary

  • Allow configuring with initial set of partitions in order to avoid starting off at 1 with empty database

• Can also have a fixed number of partitions per node and dynamically grow/shrink partitions as data increases with fixed number of nodes and fixed number is randomly chosen when new node joins cluster

Operations: Automatic or Manual Rebalancing

• Automatic rebalancing is convenient but unpredictable as it is expensive to perform so automatically commit changes but make human review in the process

Request Routing

• If we distribute the database into partitions which one do we connect and write too → service discovery and request routing
  • Can request any node w/ round robin and forward to correct node if needed
  • Can send requests to routing service which determines node to send to
  • Let clients know which partitions map to which nodes
• All the node participants need to agree else mixups happen → need to achieve consensus in a distributed system such as ZooKeeper to help

Parallel Query Execution

• For analytics queries with multiple operations, can execute this in parallel by breaking them down into multiple execution stages

Chapter 7: Transactions

• Database software/hardware or the network or concurrent operations may introduce faults in the system and we have to manage this
  • Use transactions to solve this where you group multiple reads and writes together in a unit/operation (commit or rollback) for easier error handling
• Transactions make things easier for applications and give guarantees to applications for certain requirements
  • Can relax transaction guarantees for better performance/availability but weakened safety

The Slippery Concept of a Transaction

• Can have transactions that were consistent for decades with sql or use nosql which abandons transactions for speed

The Meaning of ACID

• ACID: Atomicity, Consistency, Isolation, and Durability
  • Atomicity is all or nothing so if fault occurs during atomic transaction then operations reverted
  • Consistency the data is in a certain agreed upon state
  • Isolation concurrent transactions are isolated from each other and cannot interfere with each other
  • Durability once transaction committed it will not be forgotten and has been written somewhere

Single-Object and Multi-Object Operations

• Need ACID for Single Object writes since what happens if we only send half the content and network goes down or concurrent read/write happens
  • Implement with log for crash recovery and locks on each object for isolation + atomic operations
• Multi object transactions very difficult to implement across partitions + gets in way of performance and availability
  • Useful when operations reference tables with foreign keys + denormalized documents in nosql databases + secondary index always needs to be updated
• Can have ACID or leaderless replication where user will have to fix errors in application if they occur
  • ORMs throw an exception in application
• Retrying aborted query has a few pitfalls
  • Transaction succeeds and network fails on ack
  • Transaction fails to overload makes problem worse
  • Only retry after transient errors (deadlock/network down) over permanent error(constraint violation)
  • Need to deal with side effects outside of database (don’t send email multiple times)
  • Client process fails when retrying

Weak Isolation Levels

• Concurrency issues happen when transactions touch the same data at once and hard to debug when testing since timing based bugs so databases give you transaction isolation and hide the concurrency issues from the application
  • Serializable isolation === transactions same as if you ran one after each other which has performance impacts so can use weaker levels of isolation but may cause incidents

Read Committed

• Read committed states no dirty reads (seeing data that has only been committed) and no dirty writes (overwrite data that has been committed)
• Dirty read prevention make sure to only make writes visible in transaction all at once when committing not one after another
• Dirty write prevention make sure to delay second+ writes until first write has committed/aborted in order not to overwrite data part of 1 transaction
• Implement read committed separately for reads and writes
  • Prevent dirty writes with row level locks
  • Prevent dirty reads with above option but now reads waiting on singular write query harming performance so keep old committed value and new value set by transaction for lockless reading

Snapshot Isolation and Repeatable Read

• Read committed isolation allows read skew/non-repeatable read where values change after transactions but still allowed since reading committed values

  

• This is fine for minor cases like above but imagine backups or analytic queries/integrity checks having inconsistencies database wide

  • Use snapshot isolation here where read from a certain time of database’s value
  • Use row level locks for writes and keep multiple versions of object for reads (Multi Version Concurrency Control)

  

• To ensure consistent snapshot, at start of transaction ignore any other transaction writes for the snapshot but allow them for other application queries ie ignore all uncommitted changes

• For indexes in MVCC make index point to all versions of object and index query filters out/deletes as needed or use copy on write on BTrees where only copy child to parent pages to point to new versions

  • Append only btrees every transaction is already a new root and a snapshot of db

Preventing Lost Updates

• Can have concurrent writes where 1 write is lost due to both ignoring each others changes
  • Solution is to use atomic writes internally in database (hide from application) but using ORMs you may not notice atomic operations happening
• Application can explicitly lock which objects will be updated but now prone to application errors
  • Can also go lock free and when you detect that update has been lost, retry again
• Locks and compare and set operations can’t apply on multi leader/leaderless replication due to multiple concurrent writes happening so 1 approach is to allow all writes and merge all of them in at the end

Write Skew and Phantoms

• Write skew race condition where multiple transactions cause side effect due to stale data but does not violate any major isolation principles (dirty reads/writes, lost updates)

  • Happens when different transactions read the same objects and update different parts/different objects
  • Atomic operations don’t work as multiple objects updated, can’t detect via lost updates, constraints hard to build with multiple objects ⇒ only way forward is serializable isolation or explicitly lock rows

  

• Main template that causes this is:

  • Select query for a condition/requirement is made and decide whether to proceed or not
  • If we proceed we perform an operation(insert/update/delete) that changes the initial result of the select query (phantoms)

• Can generate all possible row combinations and then lock on those rows ⇒ materializing conflicts but hard to deal with concurrency and this problem leaks into the application layer

Serializability

• Strongest isolation level is serial isolation since isolation levels are hard to understand in context, hard to see if application code is safe to run at certain isolation level, no tools to detect race conditions
• Guarantees results to be if transactions executed one after the other

Actual Serial Execution

• Can do serializable isolation by ditching concurrency and executing only 1 transaction at a time on 1 thread

• Only came to realization once RAM cheap enough to store entire database in memory + OLTP short transactions and OLAP long running can run separately

  • Don’t make transactions long else idling time affects transaction time and runs into multiple race conditions ⇒ instead make multiple smaller transactions

• Interactive transactions cause multiple delays on single thread due to waiting ⇒ use stored procedures and application submits entire transaction code to database

  • Cons include no unified language for stored procedures (different for every database), harder to manage code running in database vs application, a badly tuned database affects multiple applications
  • Instead try and use modern languages for stored procedures

  

• Can partition data in serial execution if every transaction executes in independent partition to run on multiple threads

  • If needs to access multiple partitions then needs to coordinate locking among partitions ⇒ much much slower than single partition transactions

Two-Phase Locking

• 2 phase locking has much higher restrictions on locking but prevents race conditions
  • Originally locks implemented on database to prevent dirty operations
  • Now transactions can only read when no one else is reading OR there is only reading happening ; if a write happens then exclusive access must be given
    • If want to write, then must wait on any reads/writes to be done - if want to read then must wait on any pending writes
    • Writers block other readers and readers block other writers which is the opposite of snapshot isolation
• Used by MySQL, SQL Server, DB2
• Every object in database has lock which is either in shared mode or exclusive mode
  • Shared mode lock only acquired if 0 - multiple readers present and must wait if lock holds exclusive and exclusive lock only acquired if 1 writer
  • Upgrades from shared to exclusive are available given requirements
  • Deadlock possibilities if 1 transaction holds lock indefinitely so implement deadlock detector to abort transactions with locks as needed
• Performance significantly worse with increased isolation level due to acquiring/releasing lock overhead + reduced concurrency
  • Databases don’t limit duration of transaction + transactions need to wait on other ones due to locking leading to unstable latencies and very high p95/p99 latencies
• To prevent write skew race conditions where changing an element affects another’s search query, should implement a predicate lock which is similar to a shared/exclusive lock but belongs to an object that matches a search condition of a query instead of an object
  • If wants to read/write on a condition then acquire shared/exclusive mode predicate lock and must wait if a writer is writing
• Predicate locks have bad performance due to checking for matching locks so instead use index range locking
  • Index range locking generalizes a predicate lock (room 123 between noon - 1pm to room 123 any time to any rooms)
  • If no safe index range to lock on though then may lock entire table which isn’t good for performance

Serializable Snapshot Isolation

• Currently discussed options either don’t perform well (2 phase locking) or don’t scale well (serial execution) + differing isolation levels and you may think serializable isolation and performance are a tradeoff

  • Serializable snapshot isolation aims to solve this with full serializability but slightly lesser performance than snapshot isolation

• Pessimistic concurrency control based on the fact that something may go wrong so wait until it is safe again 2PL ⇒ serial execution most pessimistic

• Optimistic concurrency control don’t block but let it happen and hope everything turns out correctly and abort transactions that did not execute serially

  • If high contention then many transactions need to abort but ok amount of contention and system can handle throughput then fine + some operations can be commutative (counter increment) which helps

• Write skew a premise/query result can be changed based on the resulting update based on that premise ⇒ need to detect when transactions are using outdated premises and abort transactions with either detecting stale MVCC reads or detecting writes that affect prior reads

• Detect stale MVCC reads by checking if any ignored transaction writes have been committed and if so abort

  • Wait until committing to do this check because database doesn’t know if transaction will do a write + the overwritten values present may be aborted and transaction allowed to proceed

  

• To detect writes that affect prior reads, keep a record of which transactions have read which locks and if an update by a previous transaction affects a current transactions read then abort

  

• Performance is dependent on real world scenarios

  • Can figure out exactly which transactions can abort with detailed tracking but tracking overhead comes in place - have less detailed tracking and more transactions may abort
  • Sometimes ok to read information thats overwritten by transaction
  • No blocking waiting for locks and writers don’t block readers and readers don’t block writers + can be distributed across multiple machines
  • Longer transactions may lead to more aborts and conflicts so SSI requires short read write transactions but long read only transactions ok

Chapter 8: The Trouble with Distributed Systems

• Many things can go wrong in distributed systems

Faults and Partial Failures

• Computer with software either works or it doesn’t ; this gets blurry when we run software running on multiple computers
• Need to deal with the fact that things are broken in some unpredictable and nondeterministic way ie partial faIlure

Cloud Computing and Supercomputing

• HPC deals with cpu intensive tasks and cloud computing works with multiple data centers and cpus connected to each other and is scalable on demand
• HPC nodes connected to each other with DMA/shared memory so if fault happens go back to previous checkpoint and run again
• For cloud computing, very bad to make the service unavailable since they are online, built and connected to other general machines vs specialized hardware, use IP/ethernet vs specialized network topologies, allow breakages, can do rolling upgrades, and everything over the network vs nodes close together

Unreliable Networks

• Shared nothing systems have their own private cpu + ram and can only communicate with each other over the network
  • Cheap, no special hardware needed, and can use in cloud computing
• Asynchronous packet networks give no guarantee what happens to packet once it is sent out since there are many things that could go wrong
  • Timeouts help here but the failure remains unknown

Network Faults in Practice

• Network faults happen everywhere from software to hardware from accidental to human related errors and we need to handle them
  • Not handling them could lead to catastrophic errors and should expect to actively handle them and recover from them in software

Detecting Faults

• Hard to detect faulty nodes
  • If the node alive but process crashed can block TCP connections
  • If process crashed but node OS alive then notify other nodes
  • Query network switches to find hardware level failures but only works if directly own management interface
  • Routers can send ICMP destination unreachable packets
• Assume that you may not get any responses at all and that if you do get a response that the receiving process may have crashed before handling the request

Timeouts and Unbounded Delays

• Long timeouts means users will have to wait more and shorter timeouts could falsely detect dead nodes and perform operations twice, place strain on underlying systems already under load, and cause cascading failures where 1 node fails another overloads and also fails and repeats
  • One way to solve is to constantly measure how long it takes for requests to travel to and back and dynamically adjust timeout delay
• If many requests are sent to 1 destination then must queue the packet and if queue is full then drop packet since congested
  • More queueing occurs by the OS if all cpu cores are used and virtual machine monitor queue when OS paused by vm so increases delays
  • TCP uses congestion control where nodes limit sending rate to not overload receiver nodes and uses timeout delays to determine packet loss
• If delayed data is worthless (live gaming/streaming/zoom calls) then use UDP over TCP as it is faster since no implemented delays and controls

Synchronous Versus Asynchronous Networks

• Why not make networks synchronous and dedicated like a telephone line w/ circuits?
  • Need to reserve set amount of bandwidth where no one else can use unlike anyone who can open up a TCP connection
  • Packet switching over circuit switching used for bursty traffic dynamic tasks that don’t have a bandwidth requirement - if bandwidth requirements are needed could use circuit switching
    • If want to do dynamic tasks over a circuit, need to guess bandwidth allocation
      • Too low then slow transfer and too high then circuit cannot be made

Unreliable Clocks

• Clocks are used to measure durations and describe points in time
• Clock timings are inconsistent as data takes time to travel over network and every machine as its own clock which could be faster or slower than other machines

Monotonic Versus Time-of-Day-Clocks

• Time of day clocks show current time and are synchronized with Network Time Protocol Server but if too far ahead then messes up time so don’t use as constantly gets unsynchronized
• Monotonic clocks used to check time intervals (not exact time) and NTP can only adjust frequency of clock moving forward not cause it to move forward/backward

Clock Synchronization and Accuracy

• Just like we deal with the network being slow and unreliable we have to deal with clocks being unreliable as they may lead to silent data losses

Relying on Synchronized Clocks

• When clocks are not synchronized in distributed databases then might not reorder writes due to last write wins

  • Writes may disappear on nodes with lagging clocks
  • LWW depends on timestamp recency but even these timestamps may be incorrect
  • Use logical clocks based on incrementing counters

  

• Even though you may be able to read the “correct” time off a clock, it usually is milliseconds to seconds off so think of it as a confidence interval of reading the time

  • Can calculate uncertainty bound but for many provided API’s, they do not give it

• For snapshot isolation we use global monotonically increasing transaction ids for databases to figure whether out to read that certain value

  • To use timestamps from synchronized clocks would be nice but clock accuracy makes this hard
  • Spanner gives clock time as confidence interval(earliest, latest), as long as this does not overlap so earliest 1 < latest 1 < earliest 2 < latest 2 then 2’s events definitely came after 1

Process Pauses

• If we have a single leader per partition model and the leader knows it is the leader by acquiring a “lease” of time from all other follower nodes, then this could get messed up if using normal clocks
  • If out of sync then window can extend or retract by a few seconds/milliseconds
  • If thread pauses and lease expires, the old leader node may still process requests afterwards
    • Should assume that nodes can be paused for an arbitrary amount of time - threading models do not help with this only for thread safe structures
• Sometimes we want guaranteed performance and don’t want delays in system ie we want the airbags of a car to deploy no matter what even if there is a garbage collection process ⇒ real time systems
  • Needs support throughout the stack via real time OS kernel, documented worst case time to execution functions, lots of testing so slow to develop and very expensive - only used in safety critical devices
• Doing garbage collection and pausing all other nodes is expensive and hard to deal with
  • Can treat this as preplanned node outages and work around this
  • Can only use garbage collection for short lived objects and restart nodes once they hit a limit so that cleanup is quick

Knowledge, Truth, and Lies

• In distributed systems we can only pass messages to other nodes and possibly figure out its state only if they respond and nothing else
  • Very open ended questions on what is T/F in a system or is something really present?
  • Solution is to define requirements and assumptions about a system and then build the system so that it meets those requirements

The Truth is Defined by the Majority

• Imagine a node that accepts all requests but doesn’t send out responses ⇒ will unfortunately be declared dead
  • Or one stopped by a garbage collection pause, declared dead, and wakes up operating like normal again
  • Need to make sure if for any “hold one thing” scenario, that it is explicitly taken care of in a distributed system
• No matter what happens always use a quorum and the majority quorum happens and a node is declared dead that way, then it is dead no matter what nothing can stop it
• For using locks/leases on a shared resource, use a technique called fencing which returns a monotonically increasing token and when performing operation, check the token server side - the resource essentially fences off bad requests

Byzantine Faults

• Fencing is good for accidental nodes that think they’re the leader but could maliciously send a high enough fencing token to circumvent server side check - DDIA assumes nodes tell the truth
  • But what if nodes lie ⇒ dealing with byzantines fault and to reach a consensus in this environment is the byzantines generals problem
    • Generals have outposts in different areas and can communicate only by messengers - but what if some messengers are evil and spread lies
• Byzantine fault tolerant system applies if system can operate even if some nodes are malicious
  • Imagine in aerospace system some nodes/memory get corrupted due to radiation - we don’t want to kill people :^)
  • Some participants may cheat/defraud others so nodes shouldn’t blindly trust other nodes
  • Does not save you from bugs unless all implementations are independent
• Assume in the book that we don’t expect this due to all nodes being controlled by organization since byzantine fault tolerance is complicated + hardware level support needed
  • For web apps could have clients do bad things so use inout validation, sanitization, output escaping to prevent sql injection and cross side scripting ⇒ use server side validation instead
• Byzantine fault tolerance will not help for detecting security vulnerabilities because if attacker has bypassed 1 node it can probably take others
• Still want to implement procedures in place for “weak” forms of lying such as bugs, hardware issues, misconfiguration

System Model and Reality

• Algorithms need to be written with minimal dependence on hardware and software so define system models in order to understand requirements
  • Synchronous model assumed every delay and error is bounded which is unrealistic
  • Partially synchronous model is like synchronous model but bounds may be exceeded for delays, errors, and pauses which is more realistic
  • Asynchronous model no timing assumptions can be made
• Need to also define faults for when nodes fail
  • Crash stop faults nodes can only fail by crashing
  • Crash recovery faults nodes can crash and maybe recover and have disk storage that is preserved on crash
  • Byzantine faults nodes can do anything
• Usually for most systems they want partially synchronous + crash recovery models

Chapter 9: Consistency and Consensus

• If we don’t want to fail the entire system and send an error message to a user then we need to make fault tolerant distributed systems
  • Need to find a general abstraction with guarantees, implement them, and let applications rely on those guarantees like transactions

Consistency Guarantees

• When you look at most databases, they’ll have 2 different versions since they have eventual consistency so writes will catch up to them later on
  • Eventual consistency has its problems as writing a value and then reading it again could lead to different values so leads to application problems + hard to debug
  • Would want to see if there are any stronger consistency models available but at the cost of worser performance or less fault tolerant

Linearizability

• Linearizability or strong consistency makes it appear that there is only 1 copy of the database and any operations applied on it are atomic + the value read should be the most up to date value

What Makes a System Linearizable

• Could start off with a definition of linearizability as one that returns the correct value within the query bounds of reads and writes
  • But reads could read different values when a write is happening so this isn’t strict enough
  • Add in the requirement that if a read returns a new value, all reads after that should return the new value
    • Utilize atomic compare and set operation for this

Relying on Linearizability

• Few use cases where linearizability is needed vs useful
  • Locking/Leader election consensus is linearizable with Apache Zookeeper + Distributed locking
  • Need a single up to date value in a system ie unique constraint, bank account balance, flight seat-map
  • Cross channel timing/race conditions cannot get mixed up
    • Lets say you upload an image and want to resize it then resize operation sent in message queue to file server to perform operation - what if the message queue request comes in before the image is initially uploaded though?

Implementing Linearizable Systems

• Single leader replication can potentially be linearizable if reads from leader or synchronously updated followers
  • Snapshot isolation is not linearizable though
• Consensus algorithms are linearizable to prevent split brain and stale replicas
• Multi leader replication is not linearizable as they concurrently process writes on other nodes and asynchronously replicate them on other nodes ⇒ leads to conflicting writes that need resolution
• Leaderless replication probably isn’t linearizable in Dynamo style databases
  • Can make linearizable though with synchronous read repair and reading latest version before sending writes at a performance penalty
    • Only works for reads and writes and not compare and set due to consensus needed

The Cost of Linearizability

• CAP theorem can only have 2/3 of Consistency, Availability, and Partitioning ⇒ Linearizable databases only can have consistency and availability to be linearizable so cannot distribute this
• Most systems are not linearizable as you need to drop performance for it
  • RAM on multicore cpu not linearizable due to individual memory caches written to first then asynchronously written out to main memory
  • Linearizability response time of read and write requests proportional to uncertainty of delays in network

Ordering Guarantees

• Ordering is an important topic to cover and has connections with linearizability and consensus

Ordering and Causality

• System is causally consistent if it obeys the order imposed by causality (lol)
• Causal Order(where there are causal or concurrent) are not total orders (where any elements in a set can be compared) since there is no way to order concurrent events
  • Linearizability implies a total order so that means no concurrent operations should be present in the database
• Linearizability is stronger than causal consistency and also implies causality - can use causal consistency instead most of the time as it does not slow down due to network delays
• To determine causality in non linearizable database can use version vectors or some way to identify dependent relationships/versions of data read across the database

Sequence Number Ordering

• Tracking all data for causal relationships leads to large overheads - can we use some sort of sequence number (not time but logical clocks) instead?

• Single leader generates sequence numbers but for multi leader generate independent sets of sequence numbers(what if 1 independent set lags behind), attach timestamps(clock skew), preallocate blocks of sequence number(lags behind) but do not establish causality

• Lamport timestamps keep pairs of (current count value, node id) with node ids as tiebreaker

  • Not version vectors that distinguish if 2 operations are concurrent/casually dependent but instead enforce a total ordering

  

• Lamports timestamps do not work when you need to determine what is next currently instead of later

  • When 2 people create same username then lamport timestamps can determine which wins after the accounts have been created which isn’t good enough since you can’t roll this back/adjust this operation
    • To check this would have to check every other node and if one node is down then cannot proceed
  • Sometimes you need to know also when the order is finalized ⇒ total order broadcast

Total Order Broadcast

• Single leader/cpu easy to determine order but on multiple, getting nodes to agree on an order is tricky ⇒ total order broadcast/atomic broadcast ~= consensus
  • Requires reliable delivery (no messages are lost) and total ordered delivery (messages are delivered to nodes in the same order)
  • Works for database replication for consistent replicas, serializable transactions, fixed order messages, locking service with fencing tokens
  • Kind of like write ahead logs but for messages in a distributed system since all machines must process the messages in the same order
• Can implement linearizable storage with total order broadcast but they are not the same
  • Total order broadcast is asynchronous and linearizability is a recency guarantee that tells you that a read will see the latest value written
  • To implement linearizable storage with this, append message with log request, read the log and wait for message appended to be delivered back to you, if the result is own message success else abort
    • Every username value beforehand has a linearizable register beforehand
    • Works as log entries delivered to all nodes in same order and if concurrent writes then all agree on which ones come first
    • Does not work for linearizable reads as data could be stale
  • For linearizable reads, can write to log and perform read when log returned to you w/ helper function as needed to wait for latest log message entries OR read from replica that is synchronously updated on writes
• Can implement total order broadcast using linearizable storage and vice versa above
  • Even easier here as you have one linearizable register that returns a increment and get operation and all messages being sent go to that register and receive a monotonically increasing unique sequence number
    • Lamport timestamps have gaps while this doesn’t so nodes know when to wait (on 4 wait for 5 if received 6) ⇒ total order broadcast and timestamp ordering

Distributed Transactions and Consensus

• Consensus goal is to get several nodes to agree to something but many mistakes have been made building on top of it since it is so misunderstood
• Many situations need consensus such as leader election and atomic commits

Atomic Commit and Two-Phase Commit

• Atomicity prevents failed transactions from affecting database either all commit or all rolled back within 1 transaction

• Atomic commits differ for single vs multi database node transactions

  • Single database node when committing, append commit log to disk which tells you on crash if you should commit (present log) or abort (no log present)
    • So data then the log gets written when dealing with 1 database node
  • Multi database node cannot write all the commit logs due to various reasons
    • Some nodes could detect violations and need to abort
    • Some requests could get lost
    • Some nodes could crash before record is written and rollback on recovery

• When a node commits a transaction its assumed to be irreversible as other readers/writers depend on that current state of the data - need to be very careful in deciding when to commit

• Two Phase Commit algorithm that makes sure all transactions atomically commit across multiple nodes

  

  • Introduce a coordinator which manages this
  • Database serves reads and writes as usual with unique attached transaction id
  • Then when an application wants to commit, then 2 things happen
    • Prepare request sent by coordinator and expects all yes (at this point participants MUST make sure they can commit no going back)
      • Commit request sent to all nodes and commit log written in coordinator log
    • If one node says no ⇒ coordinator sends abort to all nodes
  • Coordinator will send the commit or abort infinite times until it reaches all nodes to enforce the decision
  • 2 Points of no return for Two Phase Commit: when participant commits and when coordinator receives all commit requests and sends to participants to commit
  • Coordinator is the single atomic node that the system is reliant upon - if the coordinator fails on prepare requests then participants can abort - if the coordinator fails on commit requests than all nodes must wait for the coordinator to respond
    • Must store the transaction log due to this scenario so coordinator can pick up on recovery

• Three Phase Commit aims to solve the problem of nodes in 2PC waiting for coordinator to recover by making this process nonblocking

  • However requires network to have bounded delays and nodes with bounded response times and perfect failure detection for nodes which most systems cannot ensure these bounds and instead network delays and pauses can be unbounded

Distributed Transactions in Practice

• Distributed transactions are up to 10x slower than single node ones due to disk syncing due to crash recovery + network round trips but there needs to be distinctions within these
  • Database internal distributed transactions deal with distributed transactions among the same database instances ⇒ can optimize based on software/hardware and work well
  • Heterogeneous distributed transactions dealing with multiple different database/service types ⇒ needs atomic commit and lowers performance a lot
• Heterogeneous Distributed Transactions allow any type of systems to coordinate and atomically commit any generic transaction type but all must be able to use the same atomic commit protocol
• XA (eXtended Architecture) is generic way to implement Heterogeneous Distributed Transactions supported by most relational databases and message brokers
  • C/Java/etc API for interfacing with transaction coordinator and assumes that applications use network driver or client library to communicate with participant databases/message services
  • Driver calls XA API’s to see if operation should be part of distributed transaction and sends information to database + exposes callbacks where coordinator asks participant to prepare/commit/abort
  • Coordinator implements the XA API and is a library loaded onto the process of the application issuing transactions - if application goes down then coordinator does too
• The reason we make a big deal about transactions being stuck waiting for coordinator is because we cant do other work while waiting for a response - we hold row level write locks to prevent dirty writes and row level read locks additionally for serializable isolation
  • If coordinator down for 20m then locks held on participants for 20m - if coordinator loses log then locks held forever
• If coordinator goes down and log is lost/corrupted then database admin needs to manually look at transactions and decide whether to commit or roll back and apply same results to every participant
  • Cannot restart automatically since participants will still hold locks forever due to 2PC
  • XA implementations have heuristic decisions which is escape hatch to say if we should all abort/commit in doubt transactions but this breaks atomicity
• There are many limitations of distributed transactions with XA
  • Coordinators if not replicated can be a single point of failure (hard to replicate or don’t do it at all for most systems)
  • Application servers no longer stateless with coordinator inclusion so can’t add/remove these at will anymore
  • Cannot detect deadlock and cannot do Serializable Snapshot Isolation due to how generic it is and how many systems can interact with it
  • Even for database internal transactions still rely on waiting for coordinator/all participants to respond at multiple stages else everything breaks ⇒ can we build something fault tolerant?

Fault-Tolerant Consensus

• Consensus algorithm proposes 1+ values and decides on one value and must satisfy these properties
  • Uniform Agreement: No 2 nodes decide differently
  • Integrity: No node decides twice
  • Validity: If node decides value v then v was proposed by some node
  • Termination: Every node that does not crash eventually decides some value
• Fault tolerance only cares about satisfying the first 3 properties as you assign dictator to make all the decision - but if dictator goes down then cannot make any decisions
  • Termination property implies fault tolerance - even if nodes crash we must still make a decision and cannot be stalled assuming a majority are working
  • Assume if node goes down it never comes back
• Consensus assume that it is not Byzantine fault tolerant most of the time but can work around this
• Consensus algorithms such as Raft, Paxos, Zab exist and do the process differently
  • They decide on a sequence of values instead of 1 value which makes them total order broadcast-able ~= equivalent to doing several rounds of consensus
  • More efficient to implement total order broadcast directly over 1 at a time consensus due to efficiency (Multi Paxos)
• Single leader replication is kind of like total order broadcast one might ask? But we need to elect a leader somehow.
  • Can do it manually but human intervention needed or do automatic leader election and failover but we still need to deal with split brain and get all nodes to agree on who is the leader ⇒ consensus
  • We need consensus in order to elect a leader ⇒ these algorithms are total order broadcast algorithms ⇒ total order broadcast is like single leader replication ⇒ single leader replication requires a leader ⇒ we need a leader to elect a leader?
• To elect leaders and differentiate them to avoid split brain assign epoch numbers and say within each epoch, a leader is unique
  • If leader is dead, start new election with monotonically increasing epoch number and if leader conflict happens, higher epoch number wins
  • When leader decides anything, check that it has the highest epoch number and collect votes from a quorum of nodes and compare epochs
  • 2 rounds of voting: choose leader and vote on leader proposal and in both scenarios there is at least 1 node that voted in both so can detect if new leader has appeared or not
  • Similar to 2PC but coordinator is not elected and needs yes vote from every participant instead of a quorum majority
• There are a few downfalls of consensus
  • Consensus is like synchronous replication so you trade performance for availability
  • Consensus needs a strict majority to operate
  • Consensus assumes a fixed set of nodes that vote and dynamic membership allows changes but makes more complicated
  • Consensus relies on timeouts to detect failed nodes ⇒ significant amount of time leads to multiple nodes thinking they’re the leader ⇒ frequent leader elections degrade performance as time needs to be spent choosing new leaders
  • Consensus depending on implementation is sensitive to network problems as sometimes multiple leader switching/resigning/electing occurs

Membership and Coordination Services

• ZooKeeper described as a coordination and configuration (ie consensus) service - if its implemented like a database why implement a consensus algorithm for it?
  • An application developer usually uses it indirectly with another service like Hadoop/Kafka and these services give ZooKeeper a small amount of data which is replicated across all nodes with a fault tolerant total broadcast algorithm
  • Also contains linearizable atomic operations - can implement a distributed lock ie a lease on an operation
  • Also contains a total order of operations - provides fencing tokens in order to protect a service
  • Also provides failure detections via heartbeats and timeouts and automatically cleans up resources afterwards
  • Also gives clients change notifications when new clients enter/exit the cluster
• If there are many nodes, trying to do consensus on all of them expensive so ZooKeeper does consensus on a fixed smaller amount of nodes
• ZooKeeper stores slow moving data ie node address, role, partition state and state changes occur in the scale of minutes - hours so don’t use for fast changing application data
• ZooKeeper stores service discovery data in order for other services to find other services on initialization
  • Does not need consensus and can tolerate staleness and eventual consistency
  • However in service discovery can provide the leader in there ⇒ provide read only cached replicas which receive logs of all decisions of consensus algorithm asynchronously which handles non linearizable read requests
• ZooKeeper provides membership services and tells you which nodes are alive for the membership
  • Can declare nodes dead and even though they still may be alive, a consensus protocol + failure detection is used to declare dead/alive nodes

Chapter 10: Batch Processing

• There are multiple types of systems available for use
  • Services ie request/response systems where response time and availability are important
  • Batch processing takes input data and crunches out outputs over a sustained period of time that is scheduled at certain time periods where throughput is important ⇒ MapReduce go-to example of this
  • Stream processing takes input data in real time and crunches out outputs lowering latency of jobs and builds on batch processing

Batch Processing With Unix Tools

• Can build own batch processor with Unix if you wanted too

Simple Log Analysis

• Utilize unix pipes to take output of 1 result and place it in the input of the next allowing us to process gigabytes of log files ⇒ cat x | sort | uniq -c …
  • We can also write the same batch processing in a programming language but if the dataset gets too large then massively slower than Unix due to its various optimizations

The Unix Philosophy

• Unix philosophy ~= automation, rapid prototyping, incremental iteration, experimentation friendly, breaking down projects into tasks ~= Agile today
  • Ensures that every unix program does one thing well and new jobs should be built from scratch over extended
    • Input files generally treated as immutable and can debug output at any stage in the pipeline
  • Expects that outputs to one program becomes an input to another
    • Uses file descriptors for a shared interface
  • Software should be tested and tried early
  • Tools should be used to help ease programming task workload even if you have to build those tools and throw them away later

MapReduce and Distributed Filesystems

• Unix tools can only run on a single machine, MapReduce allows batch processing on thousands+ of machine
• Unix uses input/output files and MapReduce uses Hadoop Distributed File System allowing tens of thousands of machines and petabytes of data to be processed
  • Distributed architecture allows nodes to access files stored on its own machine to write to files on other machines as needed
• MapReduce allows parallel computation across multiple machines without you having to explicitly write parallel code

MapReduce Job Execution

• MapReduce has a few steps

  • Process input files and break them into records
  • Call a mapper function to extract key/value pairs from records
  • Sort the key/value pairs by key and write to intermediate files
  • Call the reduce function on intermediate files and process key/value pairs and combine them

  

• Can chain MapReduce jobs into workflows where you write and read from certain HDFS directories

Reduce-Side Joins and Grouping

• If we need to join data in a batch process, avoid querying the database for the extra records on every record as nondeterministic and expensive over the network

  • Put the joined data as part of the Map task and MapReduce will automatically join together the data via the key which is treated as a pseudo address

  

• Can have hotkeys with lots of data in MapReduce job and are very slow since MapReduce waits for the slowest reducer to finish

  • Can run sampling job to find hotkeys, send hotkeys randomly to any random reducers and any input relating to hotkey needs to be replicated to all reducers handling key to distribute workload
  • Can also specify hotkeys beforehand and deal with them separately and do Map-Side joins

Map-Side Joins

• Reducer joins are simple and no need to make assumptions about the data, but what if we can make assumptions about the data ⇒ Don’t need reducers and sorting as mappers read from input and output to files
• Can do a broadcast hash join where can fit the join input in memory (hash table or disk index) and reference it as needed
  • Further optimization to make input in memory smaller is to Partition Hash Join input by map key if applicable
    • Further optimization is if memory is sorted based on same key so can read both input files sequentially and match records with same key for Map-Side Merge Join

The Output of Batch Workflows

• Outputs of batch workflows can be for search indexes, new databases (within batch job not external)
• MapReduce follows the Unix philosophy

Comparing Hadoop to Distributed Databases

• MapReduce being released wasn’t all that special due to Massively Parallel Processing Databases doing the same thing but MPP Databases were for OLAP queries while MapReduce + HDFS was for general workloads
  • Can easily run your code over large datasets with MapReduce and build a SQL query execution engine on top of this if you would like but some people find it limiting which is why Hadoop includes OLAP and OLTP type databases
• HDFS compared to regular databases does not enforce a fixed schema, allowing for any sort of performant data dumping and letting the consumers figure it out vs a sql database enforcing schema on write for the producer
• On failure MPP databases abort the query and either let user resubmit or try again and queries run for a few minutes at most and utilize memory to avoid disk seeking cost
• On failure, MapReduce jobs can run for hours and utilize disk writes for fault tolerance + dataset is too large to fit in memory and failure is on the task level to retry
  • Faulting in MapReduce isn’t due to hardware or implementation issues - these jobs have low priority and could be preempted by a higher priority service since Google overcommits resources for better utilization and efficiency

Beyond MapReduce

• MapReduce is simple in theory but the raw API’s are not simple to use so abstractions created on top of this in order to make it easier to batch process
• Depending on the type of processing, there could be other models that are faster

Materialization of Intermediate State

• Every MapReduce job is independent and the only points of contact for a job are its inputs and outputs
• What if you know if an output of one job will only be used as an input to another job?
  • Lots of optimizations here when you compare to Unix as Unix commands are streams and don’t need to wait for the entire batch to complete, can’t currently chain reducers together because mappers are redundant sometimes, replicating intermediate state across multiple nodes could be overkill
• Dataflow engines such as Spark handle the entire workflow as one job instead of multiple independent sub-jobs so closer to Unix this way and operators are much more flexible here than MapReduce input/output
  • Have options to do the shuffle like in MapReduce, partition inputs w/o sorting, broadcast outputs to partitions, etc
  • Makes sure expensive work is performed only when needed, no unnecessary map tasks, scheduler knows about all joins and dependencies beforehand so can make cache locality optimizations, option for intermediate state in memory, treated as streams
  • Instead of writing to disk for fault tolerance, can recompute data given input partitions and operators if deterministic else should rerun entire job

Graphs and Iterative Processing

• Dataflow engines scheduled as a DAG of jobs and can implement graph algorithms in MapReduce but slow as it reads entire input and produces new output every time
• Pregel Model one vertex sends another vertex a message where a function is called per vertex similar to a reducer call and messages are passed per round + fault tolerant unlike actor model
• Fault tolerance implemented by Pregel guaranteeing that messages processed exactly once at destination vertex and checkpointing state of all vertices at end of every iteration
• Framework determines how to partition vertices to nodes and nodes are only responsible for accepting messages and sending them to a known vertex id
  • If the graph can fit in memory on a single computer, it might be faster to do everything on a single machine vs distributed batch process due to message overhead and intermediate state being bigger than original graph

High-Level API’s and Languages

• Spark and other dataflow engines have query optimizers that can determine the most optimal joins for batch processing to minimize intermediate state and possible if joins specified in a declarative way ie say what joins are required and query optimizer decides optimal path

Chapter 11: Stream Processing

• Batch processing is bounded and run at certain intervals but what if we want real time responses to our data instead of an hour or day later ⇒ stream processing

Transmitting Event Streams

• Stream processing deals with events, producers/publishers that generate events, consumers/subscribers that get notified about and process the events
  • Want to avoid polling in stream processing as higher overhead and more polling === less chance for event to appear so instead consumers should be notified of events and process them accordingly

Messaging Systems

• Can use a messaging system that producers send too allowing it to send events to multiple consumers and there are multiple design choices to consider here
  • If messaging system overloaded do we drop messages, buffer them in a queue (crash or write to disk on overload?), or restrict producers from sending messages?
  • If messaging system crashes, will we have a durability mechanism such as writing to disk for recovery?
• You can directly message consumers from the producer with TCP/UDP but the onus of message delivery and fault tolerance is on the application itself vs a message queue
• Can use message queue which is a database for handling message streams with dynamic producers/consumers connecting to it
  • Durability is moved to broker and allows unbounded queuing most of the time and will only send async messages to producer on receipt
• Message queues look similar to databases since they could have commit protocols they are fundamentally different
  • Databases are optimized for long term storage that is very big and querying while message queues get messages in/out with a limited size
• Consumers receiving messages from a message queue can be load balanced where every consumer receives a unique message or fanned out where every consumer receives every message
• In order to know if we should remove message from queue, needs an ack from consumer or message queue will redeliver
  • If load balancing + redelivery is used, potential for messages to be processed out of order

Partitioned Logs

• Databases are more permanent where everything written is expected to be durable and new clients can get a copy of database vs message queues where everything is expected to be destructive and new clients cannot see previous messages

  • Why not combine durability of databases and low latency of message queues ⇒ log based message queues

• Log based message queue producers write to end of log and consumers consume log from start and wait till end

  • Can partition logs for higher throughput and can append to partitions independently
  • Fan out supported easily as consumers independently read log without anyone affecting each other and load balancing assign partitions to nodes
    • However should note that number of nodes consuming work in a topic === number of partitions and one slow message can block subsequent ones from being processed

• If messages expensive to process and want to parallelize work without ordering use normal message queue else if message processing cheap and ordering matters used log based

  

• Think of the log based message queue as a leader, consumers as the follower, and the offset as a log sequence number

• Logs must be cleaned up from time to time as if you keep appending, you will run out of disk space - assuming gb size and mb writes that are regular can buffer messages for days/weeks

• Log based message queues implicitly use buffering to deal with pace of producers and consumer going down will not disrupt other consumers here as independent reads + can manually configure with store offsets if goes down

  • Compare with traditional message queue where need to be careful about manual configuring and deleting since other consumers can get affected

• Regular message queues destroy messages on consumption vs can run batch type processes on log style if you wanted too with providing offsets

Databases and Streams

• Just like we applied database durability to streams we can also apply streams to databases
  • Something written to database is an event ⇒ replication logs is a stream of events ⇒ if old databases see logs and apply them in the same order we get the exact same state which is another form of event streams

Keeping Systems in Sync

• There is no single system that can handle your data needs so we need to combine multiple of them and update data in various different places
  • Can do database dumps periodically but if too slow can do dual writes where we write to all data stores but this is bad due to race conditions and queries failing leading to inconsistent system state ⇒ can solve this by making the derived data a follower of the system of record leader database ⇒ change data capture

Change Data Capture

• Replication logs of a database usually aren’t publicly available since database meant to be queries so hard to make derived data, but now CDC allows you to observe changes to database and replicate them asynchronously to other systems
  • Don’t implement with triggers as fragile and slow and don’t implement with replication log as schema changes need to be handled
  • Implement with write ahead log/bin log/oplog/etc
    • Logs may need to be cleaned up so initial version should be taken from a snapshot that is locatable in the log to gather changes to the end after startup OR we should read from a compacted log where if a key is overwritten, replace it

Event Sourcing

• Instead of recording explicit database changes in the log, can record actual user events that happened in order to understand what happened and to understand intent
  • Add row to dogs table vs dog of pomeranian type added
• We need a way to take all write event logs and convert them to reads and need to store all events instead of compact them since events are a higher abstraction level
• Should distinguish vs commands/events/facts
  • Commands are validated and become events
  • When an event is generated it becomes a fact and facts always have happened even if the event has been overridden later on

State, Streams, and Immutability

• Mutable state is the result of multiple immutable events happening: an account balance contains various credits and debits, current seats contains all reservations processed
  • Thus mutable state and append only logs of immutable events are 2 sides of the same coin
  • Databases are cached versions of the log and the log is the single source of truth
• Immutability in databases is an old concept - in financial bookkeeping if an error happens, add another transaction in append only ledger instead of modifying original transaction for auditing purposes
  • In software a lot easier to debug the issue by looking at immutable logs over directly overwriting data
  • Immutable events also allow better analytics purposes: adding and removing an item from a cart leads to nothing in the cart but 2 events for user behavior here
• Can derive multiple views from the same event log as described in above sections
  • Lots of flexibility separating form in where data is written and the form from where it is read and no need to worry about complicated schema designs ⇒ Command Query Responsibility Segregation
  • Normalizing/denormalizing data debates are gone if you can translate data from write optimized event log to read optimized state
• Can be unable to read your own writes if you write to log and read from derived view
  • Can do atomic writes, distributed transactions, linearizable writes
    • A multi object transaction is because user needs to modify data in several different place or we can design an event to give a description of the user action
• Feasible to keep immutable changes forever if there aren’t frequent updates/deletions since no need to store multiple versions of the same object

Processing Streams

• Streams can be used to facilitate CDC, used to push events to users (emails/push notifications, or can be processed as input streams to produce multiple output streams
• Very similar to batch jobs as it deals with operators/jobs but streaming is unbounded so can’t sort/sort merge join and can’t start at start of stream if it crashes and recovers

Uses of Stream Processing

• Stream processing used for monitoring ie fraud detection/military tracking enemies
• Complex Event Processing allows you to specify rules to search for patterns in streams via a declarative language (like SQL), stores these rules in a processing engine in order to filter out matched events
  • Can also do full text search to detect certain news articles in an ongoing feed
• Stream analytics allow you to record certain events and measure aggregation/statistical metrics on the data over a window - can use approximation algorithms that use considerably less memory and is fine since stream processing is never approximate (Spark streaming/Apache Flink)
• Materialized views can be formed by applying stream processor changes to application state possibly from the start to currently allowing us to query the datasource more efficiently~= the application state can be treated as a materialized view here
• Should note the difference between actor frameworks and stream processing that the former deals with concurrency and distributed execution issues and the latter with data management
  • Actor communication 1:1 and temporary logs and event logs durable and can have multiple consumers
  • Actors can communicate via a graph model either way but stream processing set up in a defined DAG

Reasoning About Time

• For determining timestamp use the event time as thats when event occurred and for determining windows, can use system clock but this breaks down if there is any delay in event processing/lag
• Processing of events may be delayed due to several reasons and measuring by processing time can be flawed if messages get delayed even though one happened before another but came later or stream processor has abnormally high requests after restarting even though request rate is the same
  • To deal with stragglers for a window we can drop them or publish a correction and retract the previous output for the window for windowing by event time

• Which clock will we use?
  • If we use mobile device local clock and send newly online device events then clock could be malformed or very late so received by server clock is better in this particular scenario
  • Can record 3 timestamps: event occur time via device clock, event sent to server via device clock, event received by the server via server clock and subtract the middle from the former to get delay and add delay to former to get true time
• There are several versions of how to define a window over a certain time period
  • Tumbling windows have fixed length and every event maps to one window
  • Hopping windows have fixed length but have an overlap window - 5 minute window with 1 minute hop means 10:00-10:05 for window 1 and 10:01-10:06 for window 2… (schedule based)
  • Sliding window similar to hopping window but will only produce output when there is data to evaluate (event based)
  • Session window no fixed length and only ends when user session ends/is inactive

Stream Joins

• Just like how we can join keys on batches we can do something similar for streams but becomes harder since new events can appear anytime
• Stream-stream joins will join 2 streams based on some sort of session id and need to choose window for join since data for one stream may lag or may never come so should ignore if out of bounds
  • For fast comparison index both streams on session id and can compare when events come
• Stream-table joins is very similar to batch processing hash joins where we load a database into the stream processor and query it locally based on id (no external db due to side effects discussed in prev chapter)
  • Batch jobs input doesn’t change but stream processing does so implement CDC on the local database being a consumer
  • Good to note that join is at beginning of time for table-table vs any specified window for stream-stream
• Table-table joins need to maintain a materialized view of multiple tables which is represented by joining multiple streams in implementation - think twitter follows and tweets tables to send/delete tweet + follow/unfollowing relationship streams
• Events from separate streams can occur anytime so when should we join on and what should we join on at that point in time?
  • If ordering across streams is undetermined then join is nondeterministic so rerunning the same job may not necessarily get the same result - solve this with a uuid on the joined record which makes join deterministic but cannot do log compaction

Fault Tolerance

• In order to deal with fault tolerance, can break the stream into small blocks and treat every block as a mini batch process which implicitly gives a tumbling window equal to batch size ⇒ microbatching
  • Can also generate rolling checkpoints of state and write them to durable storage but both approaches don’t work when output leaves the stream to an external service as restarting the service would cause side effects
• To prevent the above, we need to atomically commit the stream processing of the event and the write to downstream operator/external service
• Another way to prevent the above is to only make every operation idempotent either directly or indirectly with a message offset
  • Assumes replaying messages in the same order, deterministic processing, no concurrent updates, and on node failover need to implement fencing tokens