Distributed Systems
A distributed system consists of multiple computers and
software components that communicate through a computer
network (a local network or by a wide area network).
A distributed system can consist of any number of possible
configurations, such as mainframes, workstations, personal
computers, and so on.
The computers interact with each other and share the
resources of the system to achieve a common goal.
Distributed Systems
A Distributed System is a collection of independent
computers (or nodes) that appear to the users as a single
coherent system, and work together to achieve a
common goal. These computers communicate and
coordinate their actions by passing messages over a
network.
For more details :
[Link]
networks/what-is-a-distributed-system/
Distributed Computing
Advantages: Disadvantages:
⚫ Reliability (fault tolerance) ⚫ Troubleshooting
⚫ Scalability ⚫ Less Software Support
⚫ Sharing of Resources ⚫ Network infrastructure
⚫ Flexibility ⚫ Security
⚫ Speed
⚫ Open System
⚫ Performance
CAP Theorem (Brewer’s Theorem)
The CAP Theorem is a fundamental principle in distributed
systems architecture, formulated by Eric Brewer in 2000.
It helps us understand the trade-offs that distributed databases
and services must make when designing for reliability, scalability,
and fault tolerance.
CAP stands for:
C – Consistency
A – Availability
P – Partition Tolerance
CAP Theorem
Consistency:
Consistency: (all nodes see the same data at the same
time)
Consistency - the data in the database remains
consistent after the execution of an operation.
For example after an update operation all clients see
the same data.
CAP Theorem
Availability:
Availability: (a guarantee that every request receives a
response about whether it was successful or failed)
Availability - the system is always on (service
guarantee availability), no downtime.
every node (on either side of a network partition)
must be able to respond in a reasonable amount of
time.
CAP Theorem
Partition Tolerance
Partition Tolerance (the system continues to operate
despite arbitrary message loss or failure of part of the
system)
Partition Tolerance - the system continues to function
even the communication among the servers is unreliable,
i.e. the servers may be partitioned into multiple groups
that cannot communicate with one another.
CAP Theorem
• Theoretically it is impossible to fulfill all 3 requirements.
• A distributed system can support only 2 out of the 3
characteristics:
• Designers may decide which qualities to prioritize to obtain
the best performance and reliability for their systems by
knowing the CAP theorem
CA - Single site cluster, therefore all nodes are always
in contact. When a partition occurs, the system blocks.
CA systems can only exist when there is no network
partition, i.e., in a single data center or single node
environment.
• CP - Some data may not be accessible, but the
rest is still consistent/accurate. Critical systems (like
banking) often choose CP, prioritizing correctness
over availability during failures. Google Spanner (a
CP system) used in financial apps. Most core
banking systems are designed as CP — correctness >
uptime during partition.
• AP - System is still available under partitioning,
but some of the data returned may be inaccurate.
Example : Social Media Feed (e.g., Instagram /
Twitter / Facebook)
e.g., Cassandra, DynamoDB
Different types of data
There are 3 types:
1. Structured data - Structured data is highly organized and
follows a predefined schema or format. It is typically stored
in relational databases or tabular formats, where data is
organized into rows and columns.
Characteristics:
• Fixed Schema: Data adheres to a strict schema with defined
data types, relationships, and constraints.
• Tabular Format: Organized into rows and columns, making
it easy to query and analyze.
• Stored in Relational Databases: Commonly stored in SQL-
based databases like MySQL, PostgreSQL, or SQL Server.
• Easy to Process: Can be easily processed using SQL or other
structured query tools.
2. Semi-Structured Data : Semi-structured data does not follow
a strict schema like structured data but still has some level of
organization. It often contains tags, markers, or metadata that help
define its structure.
Characteristics:
• Flexible Schema: No fixed schema, but it may have self-
describing elements like tags or keys.
• Hierarchical or Nested Structure: Often represented in
formats like JSON, XML, or YAML.
• Stored in NoSQL Databases: Commonly stored in NoSQL
databases like MongoDB, Cassandra, or Elasticsearch.
• Easier to Scale: More flexible than structured data, making it
easier to scale and adapt to changes.
Web APIs (e.g., data returned from RESTful services).
Configuration files (e.g., YAML files for Kubernetes).
3. Unstructured Data :
Unstructured data has no predefined structure or schema. It is
often raw and unorganized, making it the most challenging type
of data to process and analyze.
Characteristics:
• No Schema: No fixed format or organization.
• Diverse Formats: Can include text, images, videos, audio,
and more.
• Stored in Data Lakes or File Systems: Often stored in data
lakes (e.g., AWS S3, Azure Data Lake) or file systems.
• Requires Advanced Processing: Needs tools like natural
language processing (NLP), computer vision, or machine
learning to extract insights.
Examples:
Text Data: Emails, social media posts, word documents.
Multimedia: Images, videos, audio files.
Sensor Data: Raw data from IoT devices without metadata.
What is NoSQL?
Stands for Not Only SQL.
NoSQL is a non-relational database management systems.
NoSQL database were developed in response to a rise in the
volume of data stored about users, objects and products, the
frequency in which this data is accessed, and performance and
processing needs.
It is designed for distributed data stores where very large scale
of data storing needs (for example Google or Facebook which
collects TB of data every day for their users).
These type of data storing may not require fixed schema, avoid
join operations and typically scale horizontally.
Why NoSQL ?
In today’s time data is becoming easier to access and capture
through third parties such as Facebook, Google+ and others.
Personal user information, social graphs, geo location data,
user-generated content and machine logging data are just a few
examples where the data has been increasing exponentially.
To avail the above service properly, it is required to process
huge amount of data. Which SQL databases were never
designed.
Instead of using structured tables to store multiple related
attributes in a row, NoSQL databases use the concept of a
key/value store.
RDBMS vs NoSQL
RDBMS NoSQL
Structured and organized data Stands for Not Only SQL
Structured query language (SQL) No declarative query language
Data and its relationships are No predefined schema
stored in separate tables Key-Value pair storage, Column
DDL,DML Store, Document Store, Graph
Tight Consistency databases
ACID Transaction Eventual consistency rather ACID
property
Unstructured and unpredictable data
CAP Theorem
Prioritizes high performance, high
availability and scalability
BASE Transaction
Before NoSQL
After NoSQL
Type of NoSQL
⚫ Document Oriented Databases:
Document oriented database stores data in the form of documents.
A collection of documents.
A document can be in a JSON, BSON, XML, YAML, etc format.
Data in this model is stored inside documents.
A document is a key value collection where the key allows access to its
value.
Documents are stored into collections in order to group different kinds of
data.
Relational model Document model
Tables Collections
Rows Documents
Columns Key/value pairs
Joins not available
Example: MongoDB, Elasticsearch, Couchbase Server, CouchDB, RethinkDB,
Terrastore, MarkLogic Server etc.
Type of NoSQL
⚫ Document Oriented Databases:
Type of NoSQL
⚫ Column Oriented Databases ::
Column-oriented databases primarily work on columns
and every column is treated individually.
A column-based database (or columnar database) stores data by
columns instead of rows.
This means all values of a single column are stored together
contiguously on disk, which makes analytical queries on large
datasets very fast
Column stores can improve the performance of queries as it can
access specific column data.
High performance on aggregation queries (e.g. COUNT, SUM,
AVG, MIN, MAX).
Works on data warehouses and business intelligence, customer
relationship management (CRM), Library card catalogs etc.
Example: Hadoop/Hbase, Cassandra, Amazon SimpleDB, HPCC,
Cloudera etc.
Types of NoSQL
Consider a sales table
ID Product Region Amount
1 TV East 2000
2 Laptop West 3000
3 Phone East 1000
Row-based → Stores like: Column-based → Stores like:
(1,TV,East,2000) (2,Laptop,West,3000) (3,Phone,East,1000)
ID: [1,2,3]
Product: [TV, Laptop, Phone]
Region: [East, West, East]
Amount: [2000, 3000, 1000]
Type of NoSQL
⚫ Key-Value Databases:
In key-value database each item in the database is stored as an attribute
name (or “key”), together with its value.
It stores data as a collection of key–value pairs, just like a dictionary,
hash table, or map data structure in programming.
Key-value stores are most basic types of NoSQL databases.
Designed to handle huge amounts of data.
In the key-value storage, database stores data as hash table where each
key is unique and the value can be string, JSON, BLOB (basic large
object) etc.
Key-Value stores follows the 'Availability' and 'Partition' aspects of
CAP theorem.
Key-Value stores can be used as collections, dictionaries, associative
arrays etc.
retrieve data using the key — extremely fast, typically O(1) time
complexity.
Example: Redis, Riak, Azure Table Storage, DynamoDB, Berkeley
DB, LevelDB, FoundationDB etc.
Type of NoSQL
Key-Value Databases:
Key Value
{"name":"Alice","email":"ali
user:1001
ce@[Link]","age":25}
{"name":"Bob","email":"bob@e
user:1002
[Link]","age":30}
{"user":"Alice","login_time"
session:xyz
:"2025-10-03T09:00:00Z"}
Type of NoSQL
Graph database:
A graph database uses graph structures with nodes, edges,
and properties to represent and store data in database.
A graph databases is faster for associative data sets and
hence it’s gaining popularity these days.
Graph stores are used to store information about networks,
such as social connections.
Each node represents an entity (such as a student or
business) and each edge represents a connection or
relationship between two node
Example: HyperGraphDB, GraphBase Neo4J, WhiteDB,
Infinite Graph, BrightstarDB etc.
Type of NoSQL
⚫ Graph database:
SQL v/s NoSQL
SQL Databases NoSQL Databases
Types One type (SQL database) with Many different types including
min0r variati0ns key-value st0res, d0cument
databases, wide-c0lumn st0res,
and graph databases.
Development Devel0ped in 1970s t0 deal with Devel0ped in 2000s t0 deal
History first wave with0f data st0rage limitati0ns 0f SQL databases,
applicati0ns. particularly c0ncerning scale,
replicati0n and unstructured data
st0rage.
Examples MySQL, P0stgres, Oracle Database M0ng0DB, Cassandra, HBase,
Ne04j
Data Specific language using Select, Thr0ugh 0bject-0riented APIs
Manipulation Insert, and Update statements,
Consistency Can be c0nfigured f0r str0ng Depends 0n pr0duct. S0me
c0nsistency pr0vide str0ng c0nsistency (e.g.,
M0ng0DB) whereas 0thers 0ffer
eventual c0nsistency (e.g.,
Cassandra)
SQL v/s NoSQL
SQL Databases NoSQL Databases
Scaling Vertically, meaning a single server H0riz0ntally, meaning that t0 add
must be made increasingly p0werful capacity, a database administrat0r
in 0rder t0 deal with increased can simply add m0re c0mm0dity
demand. servers 0r cl0ud instances.
Development Mix 0f 0pen-s0urce (e.g., P0stgres, Open-s0urce
Model MySQL) and cl0sed s0urce (e.g.,
Oracle Database)
Supports Yes, updates can be c0nfigured t0 In certain circumstances and at
Transactions c0mplete entirely 0r n0t at all certain levels (e.g., d0cument level
vs. database level)
Data Storage Individual rec0rds are st0red as r0ws Varies based 0n database type.
Model in tables, with each c0lumn st0ring
a specific piece 0f data ab0ut that
rec0rd much like a spreadsheet.
Schemas Structure and data types are fixed in Typically dynamic. Rec0rds can
advance. T0 st0re inf0rmati0n ab0ut add new inf0rmati0n 0n the fly,
a new data item, the entire database and unlike SQL table r0ws,
must be altered, during which time dissimilar data can be st0red
the database must be taken 0ffline. t0gether as necessary.
The BASE System
⚫ The BASE acronym was defined by Eric Brewer, who
is also known for formulating the CAP theorem.
⚫ The CAP theorem states that a distributed computer
system cannot guarantee all of the following three
properties at the same time:
Consistency:
Availability:
Partition tolerance:
⚫ A BASE system gives up on consistency.
The BASE System Contd..
⚫ Basically Available indicates that the system does guarantee
availability, in terms of the CAP theorem.
A response doesn’t necessarily mean the data is the latest,
but you’ll get something back
⚫ Soft state indicates that the state of the system may change
over time, even without input. This is because of the eventual
consistency model.
Example : A “likes count” on a social media post may change
automatically a few seconds later as replicas sync — even if no
new likes were added.
BASE system
Eventual consistency indicates that the system
will become consistent over time, given that the
system doesn't receive input during that time.
ACID BASE
Atomic Basically Available
Consistency Soft state
Isolation Eventual consistency
Durable
NoSQL pros/cons
Advantages: Disadvantages:
⚫ High scalability ⚫ Maturity
⚫ Dynamic Schemas ⚫ Enterprise Support
⚫ Replication ⚫ Transaction Support
⚫ Auto-sharding ⚫ Expertise (Highly Skilled
⚫ Integrated Caching Programmers)
⚫ Distributed Computing
⚫ Low Cost
⚫ No complicated Relationships
Advantages of NoSQL
⚫ Scalability: NoSQL database can be scaled up easily and with minimum
effort and hence it’s well suited for today’s every increasing database need
(bit data). NoSQL database have scalable architecture, so it can efficiently
manage data and can scale up to many machines instead of costly machines
that are required while scaling using of SQL DBMS.
⚫ With dynamic schema, if we want to change the length of column, or add
new column we don’t need to change whole table data instead the new data
will be stored with the new structure without affecting the previous data
/structure. In NoSQL databases we can insertion data without a predefined
schema.
⚫ Replication provides redundancy and increases data availability. With
multiple copies of data on different database servers, replication protects a
database from the loss of a single server.
⚫ To use replication with sharding, deploy each shard as a replica set.
⚫ Sharding is the process of storing data records across multiple machines and
is MongoDB’s approach to meeting the demands of data growth.
⚫ Many NoSQL database have integrated caching mechanism, hence
frequently used data are stored in system memory as much as possible and
discarding the need for a separate caching layer.
Disadvantages of NoSQL
⚫ Maturity – NoSQL database are new and emerging technologies. Since its
under heavy development bugs, new features, keep on arising.
⚫ Enterprise Support – If system fails company must be able to get timely
support. In case of NoSQL, there are very few companies which know the
technology and hence can be a deciding factor before using NOSQL.
⚫ Transaction Support – NOSQL doesn’t support SQL transaction features
and hence for financial application SQL are still one the best in industry.
⚫ Expertise (Highly Skilled Programmers) – There are millions of people
around the world who knows RDBMS while very few people are aware of
such technology hence getting a NOSQL programmer can be difficult.
What is MongoDB?
• NoSQL, document-oriented database
• Stores data as JSON-like documents
• Flexible schema
• Scalable and widely used
• MongoDB stores data in flexible documents.
Instead of having multiple tables you can simply
keep all of your related data together. This makes
reading your data very fast
SQL vs MongoDB
Terminologies
SQL (Relational MongoDB
Example
DB) (NoSQL DB)
Database Database library
Table Collection books
{title:
"MongoDB
Row Document Guide",
author:
"John"}
Column Field title, author
_id (auto- ObjectId
Basic Terminology
• Database → Collection → Document → Fields
• Collection = Table in SQL
• Document = Row in SQL
• _id field acts as primary key
MongoDB Basic Commands
• To see all available databases, in your terminal
type show dbs
• You can change or create a new database by
typing use then the name of the database.
use library
• create a collection using
the createCollection() database method.
CRUD Operations
• Create: insertOne(), insertMany()
• Read: find(), findOne()
• Update: updateOne(), updateMany()
• Delete: deleteOne(), deleteMany()
MongoDB Insert
There are 2 methods to insert documents into a
MongoDB database.
• To insert a single document, use
the insertOne() method.
• To insert multiple documents at once, use
the insertMany() method.
• Examples:
[Link]({title:'MongoDB Basics’,
author:'Alice’})
[Link]([{title:'MongoDB Basics’,
author:'Alice’},{title:’SQL Basics’,
author:’John’}])
MongoDB Find Data
• There are 2 methods to find and select data from a
MongoDB collection, find() and findOne()
Example: [Link]()
• This method accepts a query object. If left empty, all
documents will be returned.
• To select only one document, we can use
the findOne() method.
• This method accepts a query object. If left empty, it
will return the first document it finds
MongoDB Find
• To query, or filter, data we can include a query in
our find() or findOne() methods.
• Example: [Link]({‘author’:’john’})
Projection:
• Both find methods accept a second parameter
called projection.
• This parameter is an object that describes which
fields to include in the results.
• Example: [Link]({},{title: 1 , author: 0})
• We use a 1 to include a field and 0 to exclude a field.
MongoDB Update
To update an existing document we can use
the updateOne() or updateMany() methods.
The updateOne() method will update the first document
that is found matching the provided query.
Example: [Link]({title:’MongoDB
basics’},{$set: {price : 200}})
Update the document, but if not found insert it:
[Link](
{title:’abc’},
{$set:{price: 500, publication:’O’Reilly}
},
{upsert: true}
)
MongoDB Delete
• We can delete documents by using the
methods deleteOne() or deleteMany()
• The deleteOne() method will delete the first
document that matches the query provided.
• Example: [Link]({title: ‘xyz’})
• The deleteMany() method will delete all documents
that match the query provided.
MongoDB Query Operators
Comparison
The following operators can be used in queries to
compare values:
$eq: Values are equal
$ne: Values are not equal
$gt: Value is greater than another value
$gte: Value is greater than or equal to another value
$lt: Value is less than another value
$lte: Value is less than or equal to another value
$in: Value is matched within an array
MongoDB Query Operators
Let's find out all those records in which the price of the
book is equal to or greater than 300.
Example:
[Link]({price:{$gte: 300}})
Logical Operators
The logical operators in MongoDB are used to filter
data based on expressions that evaluate to true or false
Example:
[Link]({$and:[
{publication:’xyz’},{price:{$gt:500}]})
Indexing
MongoDB Atlas comes with a full-text search engine
that can be used to search for documents in a collection.
Indexes improve search performance
• Types: single field, compound, text, geospatial
• Example:
[Link]({title:1})
Aggregation
• Framework for advanced data analysis
• Similar to SQL GROUP BY
• Example:
[Link]([
{$group:{_id:'$author', totalBooks:{$sum:1}}}
])
Aggregation Pipeline
Aggregation operations allow you to group, sort,
perform calculations, analyze data, and much more.
Aggregation pipelines can have one or more "stages".
The order of these stages are important. Each stage acts
upon the results of the previous stage.
Aggregation
[Link]([
// Stage 1: Only find documents that have price more
than 100
{
$match: {price: { $gt: 100 } }
},
// Stage 2: Group documents by publication and sum
each publication books price
{
$group: { _id: "$publication", totalSales: { $sum:
"$price" } }
}
])
Aggregation $group
Aggregation $group
• This aggregation stage groups documents by the
unique _id expression provided.
• Don't confuse this _id expression with
the _id ObjectId provided to each document.
[Link](
[ { $group : { _id : "$genre" } } ]
)
Aggregation $limit
Aggregation $limit
This aggregation stage limits the number of documents
passed to the next stage.
[Link]([ { $limit: 5 } ])
Aggregation $sort
Aggregation $sort
This aggregation stage groups sorts all documents in the specified sort order.
[Link]([
{
$sort: { “price": -1 }
},
{
$project: {
“title": 1,
“publication": 1
}
},
{
$limit: 5
}
])
Aggregation $match
Aggregation $match
This aggregation stage behaves like a find. It will filter
documents that match the query provided.
[Link]([
{ $match : { genre : “Sci-fiction" } },
{ $limit: 2 },
{ $project: {
“title": 1,
“author": 1,
“price": 0
}}
])
Aggregation $addFields
[Link]([
{
$addFields: {
avgGrade: { $avg: "$[Link]" }
}
},
{
$project: {
"name": 1,
"avgGrade": 1
}
},
{
$limit: 5
}
])
Aggregation $count
Aggregation $count
This aggregation stage counts the total amount of
documents passed from the previous stage.
[Link]([
{
$match: { "cuisine": "Chinese" }
},
{
$count: "totalChinese"
}
])
Aggregation $lookup
Aggregation $lookup
• This aggregation stage performs a left outer join to a
collection in the same database.
There are four required fields:
• from: The collection to use for lookup in the same
database
• localField: The field in the primary collection that
can be used as a unique identifier in
the from collection.
• foreignField: The field in the from collection that can
be used as a unique identifier in the primary
collection.
• as: The name of the new field that will contain the
matching documents from the from collection.
[Link]([
{
$lookup: {
from: "movies",
localField: "movie_id",
foreignField: "_id",
as: "movie_details",
},
},
{
$limit: 1
}
])
This will return the movie data along with each
Map- Reduce
• Map-reduce is a data processing paradigm for
condensing large volumes of data into
useful aggregated results.
• To perform map-reduce operations, MongoDB
provides the mapReduce database command.
• Starting in MongoDB 5.0, map-reduce is deprecated:
• Instead of map-reduce, you should use
an aggregation pipeline. Aggregation pipelines
provide better performance and usability than map-
reduce.
Map- Reduce
Problem Statement :
Return the Total Price Per Customer
Perform the map-reduce operation on
the orders collection to group by the cust_id, and
calculate the sum of the price for each cust_id:
Map-Reduce
[Link] the map function to process each input
document:
In the function, this refers to the document that the map-
reduce operation is processing.
var mapFunction1 = function() {
emit(this.cust_id, [Link]);
};
The function maps the price to the cust_id for each
document and emits the cust_id and price.
Map-Reduce
2. Define the corresponding reduce function with two
arguments keyCustId and valuesPrices
var reduceFunction1 = function(keyCustId, valuesPrices) {
return [Link](valuesPrices);
};
The valuesPrices is an array whose elements are
the price values emitted by the map function and
grouped by keyCustId.
The function reduces the valuesPrice array to the sum of
its elements.
Map-Reduce
3. Perform map-reduce on all documents in
the orders collection using the mapFunction1 map
function and the reduceFunction1 reduce function
[Link](
mapFunction1,
reduceFunction1,
{ out: "map_reduce_example" }
)
This operation outputs the results to a collection
named map_reduce_example.
If the map_reduce_example collection already exists,
the operation will replace the contents with the results of
this map-reduce operation.
Aggregation Alternative
[Link]([
{ $group: { _id: "$cust_id", value: { $sum: "$price" } } },
{ $out: "agg_alternative_1" }
])
Replication & Sharding
• Replication and sharding are two key features of
MongoDB that enhance data availability,
redundancy, and performance.
• Replication involves duplicating data across
multiple servers by ensuring high availability and
fault tolerance
• sharding distributes large datasets across several
servers to manage large volumes of data and handle
high throughput operations.
Replication
Replication in MongoDB refers to the process of
copying data across multiple servers, ensuring that
multiple copies of the same data exist at different
physical locations
At any given time, one member of the replica set acts as
the primary node, and others serve
as secondary nodes.
The set of servers that maintain the same copy of data is
known as replica servers or MongoDB instances
Replication
Key Features of Replication
Replica Sets: Replica sets consist of multiple nodes
(usually an odd number for elections) that contain
identical copies of the data.
Write and Read Operations: The primary node
handles write operations, while secondary nodes provide
read scaling by distributing read queries.
Automatic Failover: If the primary node goes down, a
secondary node is automatically promoted to primary,
ensuring continuous availability.
Oplog: A special capped collection on the primary node
that records all changes. The secondary nodes use this
log to keep themselves updated.
Advantages of Replication
High Availability: Ensures data is always available,
even during server failures.
Disaster Recovery: Multiple copies of data across
different servers provide a safety net in case of hardware
failure.
No Downtime for Maintenance: Operations like
backups, index rebuilding, and system maintenance can
be done without interrupting database operations.
Read Scaling: Multiple secondary nodes allow for load
balancing, as read operations can be distributed across
them.
Sharding
• Sharding is a method for distributing
large collection(dataset) and allocating it across
multiple servers.
• It is designed to handle horizontal scaling by
partitioning data into smaller, more manageable
pieces, which are then spread across multiple servers.
• This enables MongoDB to handle high-throughput
workloads and large datasets that cannot fit on a
single server.
Why Sharding Necessary
Sharding is especially useful when:
• The database contains huge datasets that exceed a
single server’s storage capacity.
• High traffic applications require fast query
performance across large datasets.
• Distributed database architectures are needed for
large-scale applications.
How does Sharding work?
Shard: Each shard is a replica set that holds a subset of
the data. Each shard is responsible for a portion of the
overall dataset.
Config Servers: Config servers store metadata about
the sharded cluster and manage the distribution of data
across shards.
Query Routers: These servers route client queries to
the appropriate shard based on the data distribution.
Sharding
Advantages of Sharding
• Sharding adds more server to a data field
automatically adjust data loads across various
servers.
• The number of operations each shard manage got
reduced.
• It also increases the write capacity by splitting the
write load over multiple instances.
• It gives high availability due to the deployment of
replica servers for shard and config.
• Total capacity will get increased by adding multiple
shards.
Thank
You