CM3.Rmd

---
title: "Big data tools"
subtitle: "Lesson 2 — Parallelized computing on distributed data"
author: Arthur Katossky & Rémi Pépin
institute: "ENSAI (Rennes, France)"
date: "March 2020"
output:
  xaringan::moon_reader:
    css: ["default", "css/presentation.css"]
    nature:
      ratio: 16:10
      scroll: false

---

# Forewords

---

## Course outline

1. Refresher `<= Last week`
2. File systems vs. databases
3. The fundamental problems of distribution
4. Distributing file systems
5. Distributing databases `<= Today`
6. Distributing tasks
7. Parallelizing computation on distributed data
8. Statistical applications
9. Conclusions and perspectives

---

## QCM

Go on Moodle. You have 10 mins.

---

# 1. Refresher

<!-- vocabulary: cluster, node, etc. -->

---

## Storage improvement and limits

???

Storage has improved a lot, but not as fast as data size.
We generate much more data than we can store on a single hard disk.

---

.height600[![](img/storage-cost.png)]

---

.height600[![](img/storage-average.png)]


---
## Parallelisation

???

Parallelisation is not the ideal solution, it most of the time does not bring as much improvement of a program as you might expect, since some parts of a programmed cannot be parallelised. When you parallelise, a lot of time is pent moving data around.

The map-then-reduce scheme is a clever and genral way to parallelise algorithms.

---
## Cloud computing

???

Cloud computing allows you to operate much of your computing tasks in the cloud.
In particular, you can store data too big to be stored on your personal computer.
You can launch a full cluster of machines, and arrange them as you wish, to do whatever tasks you want.
Also, they provide readily available solutions for storage and for computation at large scales.
However, this storage solutions hides a lot of the complexity that it is to store data on several computers.

---

# 2. File systems vs. databases

---

## File systems vs. databases

**Goal:** what do we do when data exceeds the physical limits of our storage?

-> We first have to understand how things are stored in the normal case.

???

Last time we did not really talk about either file systems or databases.
This will be the focus of this course.

---

## File systems

- Manages files on storage space (persistant memory, i.e. most often hard disk)
  - Makes the "physical file" (bits) and the "logical file" (fragments) transparent for the user
  - Does not care about file formats
  - Can handle _heterogeneous_, _unstructured_ data (image, sound, text, application)

- Principal use cases :
  - Read
  - Write
  - Security (access rights)

???

Transparent here means "not visible", "not disturbing".


---

## File systems

### Storage schema

.height400.center[![](img/fragments.jpg)]



---

## File systems

### Traditionnal file processing

.height400.center[![](img/filesystem.jpg)]


---
## File Systems

### Key features

- Is responsible for the correspondance between the physical and logical file
- Maintains the **index** of where data is physicaly situated on disk
- Maintains the **namespace**, a virtual hierarchy of files and directory making it possible to identify one single file by a string, called a **path**
- Doesn't undestand file contents, only understands meta data (e.g. size, owner, time stamp of last change)
- Responsible of the integrity of files
- Responsible of the redundancy of files (_not needed on your personal computer_)
- Security (access rights)

**Provides an useful abstraction for the clients (=programmes)**. It's not the programm responsability to know how files are stored (physical location, redundancy, access right). But it's the programm responsability to know how to read files.

???

All files are just bytes for it, the FS open a stream between the file and the application which want to read it. The application know how to read it.

---

## Data bases

- Manages files on storage space (persistant memory, i.e. most often hard disk)
  - Makes the "physical file" (bits) and the "logical file" (fragments) transparent for the user
  - Possibly care about data format (numbers, texts, dates...)
  - Can handle _homogeneous_, _structured_ data

- Principal use cases :
  - Unique entry point to access data
  - Knows how to read/write data
  - Exposes new tools to manipulate data
      - Specific lanquage (SQL for instance)
      - Transactions and concurrency management
      - Models to organize data

???

We mix the database (storage part) and the DBMS (management system). It's not a mistake, just a simplification.

---

## Data bases

### Why use a data base ?

- Integrated tool to process data
- "Centralize" the data

---

## Data bases

### Data base processsing

.height400.center[![](img/database.jpg)]

---

## Data bases

### Data base transaction

- Data base responsibility to guarentee the coherence and validity of data
- **Transaction** : unit of work
  - **A**tomicity : complete entirely or not at all
  - **C**onsistency : changes affected data only in allowed ways
  - **I**solation : must no affect other transcations
  - **D**urability : changes must be written on persistant storage 

---

background-image: url(https://media2.giphy.com/media/5C472t1RGNuq4/giphy.gif?cid=790b76114b58b79ee92f2fa959969b9d111633782b5a5a7a&rid=giphy.gif)
background-size: cover

---

# 3. The fundamental problems of distribution

---

## Are the problems fundamentally different for file systems and databases?

The distinction becomes blur, especially when you start giving up on the _relational_ aspect of databases.

> The difference between a distributed file system and a distributed [database] is that a distributed file system allows files to be accessed using the same interfaces and semantics as local files.

Such semantics include using folder-like hierarchy, attributing permissions to individual files, copying files to different locations, etc.

We will thus start by the _common_ problems.

---

## The fundamental problems of distribution


- **Availability:** you want your data available 24/7

--
  
  - **Latency**: you don't want to wait for hours to get answered (read and write)
  
--
  - **Throughput:** maybe you want to read ALL your data or write GB at once
  
--
  - **Fault tolerance:** a single node failure shouldn't make your data unavailable

--

- **Coherence:** the same query with the same data return the same result

--
  - **Atomicity*:** complete entirely or not at all
  
--
  - **Durability*:** your data mustn't get corrupted over time
  
--
  - **Inter-node consistency:** at the same time, all the node see the same data

--

- **Schema consistency*:** changes affected data only in allowed ways


--

- **Isolation*:** transactions mustn't overlap

--

- **Elasticity/Scalability:**(under constraint of constant / acceptably-increasing request time)

--
  - can it store more or bigger files ?
  
--
  - can it process more requests ? (reads / writes)
  
--
  - can I add more nodes ? (_horizontal scaling_)
  
--
  - does scaling require a lot of _ad hoc_ work? can the scaling happen automatically? (_elasticity_)


???

* are ACID properties
---
## The fundamental problems of distribution


Plus the usual questions of large-scale systems :

- **Confidentiality:** only authorized person can access your data
- **Data gouvernance:** under which law are your data ?
- **Environment:** is your system not too big ? Does your system use too much energy ?
- **Economy:** isn't it too expensive to manage?
- ...

---

## The fundamental problems of distribution

### Some solutions 

- **Redundancy / replication:** keep copies of the data in far away nodes, so that you don't lose information under hardware failure (.green[++ availability], .green[++ fault-tolerance], .green[++ durability], .red[-- inter-node consistency], .red[-- schema-consistency], .red[--cost], .red[--environment])

--
- **Balancing\rebalancing:** use all your node fairly (.red[--availability now], .green[++ availability later], .green[++ scalability])

--
- **Timestamp-based concurrency control:** use timestamp to resolve conflict (first in first out)  (.green[++isolation], .red[--availability])

--
- **Get the closest data to the client:** if the data are close to the client, there is less network time (.green[++ availability], .red[-inter-node consistency], .red[-governance]) (*edge, fog, mist computing*)

--
- **Have a master:** it organizes the work to avoid conflict (.green[+consistency], .red[-availability], .red[-fault-tolerance])

--
- **Asynchronous processing:** nodes can accept change locally, and consolidate the transactions only in a second phase (.green[++availability], .red[--inter-node consistency])

--
- **First-class actions:** you may chose to priviledge reads over writes, or to completely prevent modifying files, for instance (.green[++availability], .green[++consistency])

???


In this part we do not distinguish between file systems and databases.

Homegeneous (all run with the same sowftware / OS) vs. inhomogeneous (diff. software / OS).


> Confidentiality, availability and integrity are the main keys for a secure system.

> A server belongs to a rack, a room, a data center, a country, and a continent, in order to precisely identify its geographical location
> The need to support append operations and allow file contents to be visible even while a file is being written
> Communication is reliable among working machines: TCP/IP is used with a remote procedure call RPC communication abstraction. TCP allows the client to know almost immediately when there is a problem and a need to make a new connection.

**Source:** https://en.wikipedia.org/wiki/Distributed_file_system_for_cloud


> Distributed file systems may aim for "transparency" in a number of aspects. That is, they aim to be "invisible" to client programs, which "see" a system which is similar to a local file system. Behind the scenes, the distributed file system handles locating files, transporting data, and potentially providing other features listed below.
> 


---
## The CAP theorem

![](img\CAP theorem.jpg)

???

Brewer's theorem, published as a conjecture in 1999, proved in 2002
Source : https://en.wikipedia.org/wiki/CAP_theorem

---
## Transparency requirements
- **Access transparency:** clients are unaware that files/data are distributed and can access them in the same way as local files are accessed

.footnote[**Souce:** https://en.wikipedia.org/wiki/Clustered_file_system#Distributed_file_systems]

--
- **Location transparency:** the way to refer to a file/data does not depend on its location

--
- **Concurrency transparency:** all clients should have the same view of the state of the file system / database

--
- **Failure transparency:** clients should not notice a single node failure

--
- **Heterogeneity and scale transparency:** it should not matter on which specific machines or on how many machines the file system / the database is distributed

--
- **Replication transparency:** clients should be unaware of the file replication performed across multiple servers to support scalability

--
- **Migration transparency:** files should be able to move between different servers without the client's knowledge (*interoperability*).

--
- ...

---
# 4. Distributed file systems

???

**Source:**

- https://en.wikipedia.org/wiki/Clustered_file_system#Distributed_file_systems
- https://en.wikipedia.org/wiki/Distributed_file_system_for_cloud

---

## Distributed file systems

**Distributed file systems** are also known historically as **network file systems**

Dates back from the beginning of "the network" in the 1960s.

In 1985, Sun Microsystems creates "Network File System" (NFS), still in use today.

---

## An exemple of distributed file system: HDFS

HDFS stands for "Hadoop Distributed File System."

It is an open-source project financed by the Apache Fundation.

.footnote[

![](img/hadoop.png)

This section is heavily inspired from HDFS documentation pages ([link](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html)). Asterisks (*) denote (almost) exact citations.

]

???

**Sources:**

- https://en.wikipedia.org/wiki/Apache_Hadoop (the explanation seems to be about Hadoop v1 since HDFS is refered to as the job tracker)
- https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html

---

## An exemple of distributed file system: HDFS

_Hadoop_ is actually a complete suite of _modules_, of which HDFS and YARN are the basic components.

But "_Hadoop_" in a broader sense refers to a complete software ecosystem, most of which is also supported by the Apache fundation. This ecosystem encompasses the _Hodoop_ modules MapReduce, Ozone and Submarine, and the libraries Ambari, Avro, Cassandra, Chukwa, HBase, Hive, Mahout, Pig, Spark, Tez and ZooKeeper. More information on [Hadoop's website](https://hadoop.apache.org).

---

![](img/hadoop-ecosystem.png)

.footnote[


**Source:** https://www.oreilly.com/library/view/apache-hive-essentials/9781788995092/e846ea02-6894-45c9-983a-03875076bb5b.xhtml

]

???

<!-- CHECK : what do the libraries mentionned on the previous slide do, that are not displayed on this table (e.g. Tez)  ? Conversely, what are the software mentionned in this table that are not in the previous list (e.g. Flume) ? -->

---

## An exemple of distributed file system: HDFS

_Hadoop_ was first released in 2006, and has evolved a lot since.

HDFS's developement was inspired by the publication of _Google File System_, a now deprecated¹ file system developped by Google, to serve as infrastructure for the _Apache Nutch_ web search engine project.

We will here focus on the latest version, Hadoop 3.

.footnote[_Hadoop_ is mostly coded in _Java_. <br/>
¹source :https://www.systutorials.com/colossus-successor-to-google-file-system-gfs/ <br/>
map-reduce principles : https://static.googleusercontent.com/media/research.google.com/fr//archive/mapreduce-osdi04.pdf <br/>
Google File System : https://static.googleusercontent.com/media/research.google.com/fr//archive/gfs-sosp2003.pdf]

---

## An exemple of distributed file system: HDFS

### Why HDFS?

- open-source

- well-documented

- well-spread

---

## An exemple of distributed file system: HDFS

### The architecture

HDFS has a master/slave architecture.

**NameNode:** a master server that manages the file system namespace and regulates access to files by clients*

**DataNodes:** slaves which manage storage attached to the nodes that they run on*

---

## An exemple of distributed file system: HDFS

### Key ideas

When given a new file, the **NameNode** splits it into one or more **blocks** (default size 128MB) and gives these blocks to be stored into a set of **DataNodes**. Each block is stored several times, on different DataNodes, and this number is the **replication factor** of that file.

**Files in HDFS cannot be modified**, except for appends and truncates. Emphasis is indeed on the reading part, a scheme HDFS calls "write-once-read-many".

By default, the replication factor is 3, and the NameNode tries to allocate the replicas intelligently: one on a given node, that then sends a copy to a close node (faster but less fault-tolerant) and an other to a further away node (slower but more fault-tolerant).

---

## An exemple of distributed file system: HDFS

### Key ideas

If a client wants to write a file, they ask the NameNode.

1. The NameNode splits the file into blocks.

2. For each block, it selects a number of DataNode to write onto (typically 3), based on:
    1. disk space (more space is better)
    2. proximity to the client (closer is better)
    3. proximity to each other (one replication close, one far)
    4. distribution of blocks (blocks of the same file should be on different nodes)
    
3. It then passes the block split and the lists to the client.

4. The client writes each block on the first block on the list.

5. The DataNode passes the block and the list on to the next DataNode on the list. (This minimizes the use of the network in between the cluster and the client, likely to be slower than the network inside the cluster.)

.footnote[The "client" is often not directly the user, but some other module asking for read / write access.]

---

## An exemple of distributed file system: HDFS

### Key ideas

If a client wants to read a file, they ask the NameNode.

1. The NameNode looks into the index to associate the file path to blocks.
2. The NameNode locates DataNodes containing the blocks, closest to the client.
3. It passes the information to the client, who in turn reads directly from the specified DataNodes.

.footnote[The "client" is often not directly the user, but some other module asking for read / write access.]


---

## An exemple of distributed file system: HDFS

### Key ideas

.center.height400[![](img/hadoop example.jpg)]

---



## An exemple of distributed file system: HDFS

### Role of the master (NameNode)

The **NameNode** :

- is the entry point for a client's requests
- decides how to split files into blocks and on which DataNodes to store these blocks
- never handles actual files, only stores metadata
- knows at any time the correspondance between a file's name and the block's identifiers (aka **namespace**, stored on disk) and the the physical location of blocks (aka **block index**, kept in memory)
- detects DataNode failure by listenning to their **heartbeat** and demands block replication as necessary


???

It periodically receives a Heartbeat and a Blockreport from each of the DataNodes in the cluster. Receipt of a Heartbeat implies that the DataNode is functioning properly. A Blockreport contains a list of all blocks on a DataNode.
- splits 
- manages and stores the data registry


The NameNode is the arbitrator and repository for all HDFS metadata. The system is designed in such a way that user data never flows through the NameNode.*

The number of copies of a file is called the replication factor of that file. This information is stored by the NameNode.*

---

## An exemple of distributed file system: HDFS

### Role of the master (NameNode)

The NameNode executes file system namespace operations like opening, closing, and renaming files and directories. It also determines the mapping of blocks to DataNodes.*

The **namespace** is persisted to disk at regular intervals (every X seconds or every Y changes). In between, a record of the changes is also written onto disk, in a file know as the edit log, so that at all time, the entire file system is preserved. In a case of a failure of the master, the data on disk is restored.

_Where_ exactly the blocks are stored, however, is not stored on disk, but kept in memory.

---

## An exemple of distributed file system: HDFS

### Role of the slaves

The DataNodes are responsible for serving read and write requests from the file system’s clients. The DataNodes also perform block creation, deletion, and replication upon instruction from the NameNode.*

**DataNodes:**
- emit a regular **heartbeat** containing the list of all the blocks stored locally
- send copies directly to each other when the NameNode requires a copy to be made
- give access (in read or write) directly to the client

---

## An exemple of distributed file system: HDFS

### Properties

#### Fault-tolerance

> The primary objective of HDFS is to store data reliably even in the presence of failures. The three common types of failures are NameNode failures, DataNode failures and network partitions.

---

## An exemple of distributed file system: HDFS

### Properties

#### Fault-tolerance

1. Hadoop modules have **location awareness**, i.e. they use the locations of the nodes relative to each other. This enables HDFS to obtain **safe redundancy** by replicating data in different locations. By default, there is one copy in vicinity (typically in a server room, the same rack) and one copy in further away (typically, an other rack).

---

## An exemple of distributed file system: HDFS

### Properties

#### Fault-tolerance

2. Failure is explicitly taken into account in the design of the file system. There may be 3 kinds of failure:
  
  - **NameNode failure:** when the NameNode fails, it is restarted, and:
    
      - restores the latest index saved on disk,
      - applies all the changes that are recorded in the edit log, also saved on disk
      - reconstitute the location of blocks in memory from the DataNodes' heartbeat
    
    Running several NameNodes with a distritbuted edit log is also possible.
    
  - **Network failure:** when a subset of DataNodes lose connectivity with the NameNode (aka _network partition_), the NameNode detects the absence of _heartbeat_ and immediately demands replication of blocks below the replication factor. Since replication happens in distinct locations, it is unlikely that all of the replicated blocks become unavailable. (By default, a DataNode is considered dead after 10 min of silence.)
  - **DataNode failure:** this is a special case of _network partition_ with only one node disconnected from the rest

---

## An exemple of distributed file system: HDFS

### Properties

#### Availability

1. **Latency / throughput** is achieved by delegating to the slave all the communication intensive work. Yet, the NameNode is a bottleneck, since all communication goes through it. **High throughput** is obtained at the expense of a (relatively) **low latency**. HDFS is not conceived for interactive use.

2. HDFS supports natively **balancing**: it will use first the least used resources.

3. HDFS is also compatible with _rebalancing_.

???

In some cases, user requests may start to concentrate on only a few NameNodes, and cause over-load and slow-down. Maybe a specific file is requested particularly often. Or maybe some DataNodes are full, whereas some other are almost empty — this might happen because you just added a new node, or because you deleted some voluminous file.  (i.e. moving data blocks from over-used nodes to under-used ones, or replicating often-requested blocks), even though it is not implemented by default as of today.

---

## An exemple of distributed file system: HDFS

### Properties

#### Scalability

1. By splitting files into blocks, HDFS does not limit the size of a single file, nor the number of files.
2. You can add slave servers at any time.
3. Users interact only briefly with the NameNode, and most of the interaction with the cluster is decentralized, hence enabling a large amount of simultaneous users.

However, HDFS is not meant for frequent writes, and does not scale in this regard. Indeed, even though data does not go through the DataNode, editting the index — which is, please remember, copied onto disk — and demanding copies to be made to the DataNodes take time. Multiple NameNode can cope with this but is not the standard installation. HDFS calls it "write-once-read-many".

---

## An exemple of distributed file system: HDFS

### Properties

#### Integrity

HDFS insures integrity by storing the _**checksum**_ of a block alongside the block itself. If the block gets corrupted, the computed checksum does not match the stored one. (It is possible, but much less likely, that the checksum itselft gets corrupted.)

Clients checksum the block they download and inform the DataNode of the result. DataNodes also run regular checsksum of all unchecked blocks.

DataNodes store information about the checks already made, and thus can narrow the date about which data has become corrupted.

In case of checksum mismatch, clients flag the incriminated block as corrupted to the NameNode, in which case a new copy is demanded.

---
## Other distributed file systems

- Proprietary private file systems such Amazon S3, Azure Storage, Google Cloud Storage, etc.
- **NFS** (Linux), **SMB** (Windows) or **AFP** (Apple), typically used in Network-attached storages (NAS's)
- **Lustre**, typically used in clusters of super computers
- ...

???

**Source:**https://en.wikipedia.org/wiki/Comparison_of_distributed_file_systems 

<!--Highlight some differences?-->

> Network File System (NFS) uses a client-server architecture, which allows sharing files between a number of machines on a network as if they were located locally, providing a standardized view. The NFS protocol allows heterogeneous clients' processes, probably running on different machines and under different operating systems, to access files on a distant server, ignoring the actual location of files. Relying on a single server results in the NFS protocol suffering from potentially low availability and poor scalability. Using multiple servers does not solve the availability problem since each server is working independently.[5] The model of NFS is a remote file service. This model is also called the remote access model, which is in contrast with the upload/download model:
> Remote access model: Provides transparency, the client has access to a file. He send requests to the remote file (while the file remains on the server).[6]
> Upload/download model: The client can access the file only locally. It means that the client has to download the file, make modifications, and upload it again, to be used by others' clients.
> The file system used by NFS is almost the same as the one used by Unix systems. Files are hierarchically organized into a naming graph in which directories and files are represented by nodes.

---
# 5. Distributed databases

.footnote[This section is heavily inspired from « Cassandra: The Definitive Guide: Distributed Data at Web Scale. » by Jeff Carpenter]

???

https://en.wikipedia.org/wiki/Distributed_database

---
## Specific problems of distributed databases

A key feature of (relational) databases is their **consistency**.

For instance, if your schema requires an foreign key, you are not allowed to add a line to a given table without linking to a record from an other table.

However, when distributing databases, it becomes increasingly difficult to maintain **consistency** accross copies of many tables. Indeed, the mere lookup of a foreign key in a table distributed across many nodes may take a long time, and one is supposed to do so for each record created / updated.

In order to maintain **consistency**, databases thus generally give up the relational component, and are thus known as **no-SQL databases**.

Doing so, they become mere **distributed data stores**, and **the frontier with a file system tends to blur**. They often consist in just a requesting programme and an interface with the underlying file system.

---
## Specific problems of distributed databases

More fundamentally, **strict consistency** means every node containing the data gets updated simultaneously ; this means that:

- there is some global time that the nodes can refer to ;
- that they can agree in a sure way that some change will be acknowledged at a given instant (and that the other nodes know that this node knows that hey know it knows... ¹) ;

The only way to insure this form of consistency is through locking-releasing resources.

.footnote[¹ This is known as **the Byzantine Generals Problem**, where to generals refuse to attack until they get completely sure that the other is also sure that they both agree to attack.]



---
## Specific solutions for distributed databases

### Denormalisation

.center.height400[![](img/join db.jpg)]



---
## Specific solutions for distributed databases

### Denormalisation

When data becomes large, join operations becomes expensive. (So do consistency checks on foreign keys.)

A solution is to pre-compute joins, so that the lookup is performed only once. This is known as denormalisation.

But then maintaining consistency accross the tables become a problem.



---
## Specific solutions for distributed databases

### 2 phase-commit

In order to maintain schema-consistency, you lock a resource until you finished the transaction.

Downsides:

- The resource is locked in between, preventing other clients to access it. It is thus only acceptable for transactions occuring super fast.
- Infinite waiting is possible.

---
## Specific solutions for distributed databases

### Compensation

In order to maintain schema-consistency, you undo a previously done change, in case that the transaction finally ends up with an error. Like soustracting -10 if you initially did +10.

Downsides:

- does not work for legally binding transactions (stock sales, bank transfers...)

---
## Specific solutions for distributed databases

### Sharding

<!-- read more -->

Sharding is the idea to split data over multiple nodes.

1. functionnal segmentation
2. (manual) key-based sharding
3. lookup-table

Downsides:

- inbalance / no duplication
- single point of failure
- every request goes through the lookup table
- sharding has to be cleverly chosen

<!--

## Specific solutions for distributed databases

### _share-nothing_ architecture

A _share-nothing_ architecture is an architecure where each node is independant from each other, so that consistency becomes irrelevant.

.footnote[Michael Stonebraker, 1986, "The Case for Shared Nothing."]
-->

---
## Specific solutions for distributed databases

### Sharding

.pull-left[

Product table, store on one node

| Product | Price |
|---------|-------|
| p1      | 45    |
| p2      | 50    |
| p3      | 200   |
| p4      | 230   |
| p5      | 500   |
| p6      | 12    |
| p7      | 56    |
| p8      | 1000  |
]

.pull-right[
Product table sharded. Each shard is stored on one node (or multiple for redundancy)

.pull-left[
Shard 1, price < 100

| Product | Price |
|---------|-------|
| p1      | 45    |
| p2      | 50    |
| p6      | 12    |
| p7      | 56    |
]
.pull-right[
Shard 1, price >= 100

| Product | Price |
|---------|-------|
| p3      | 200   |
| p4      | 230   |
| p5      | 500   |
| p8      | 1000  |
]
]

---
## Specific solutions for distributed databases

### No-SQL databases

If we accept some departure from the ACID requirement, it is possible to reach (horizontal) scalability and high availability though distribution.

This has always existed, but the unifying "no-SQL" labels was a 2010s phenomenon.

- key-value store (ex: Voldemort)
- column store (ex: Cassandra)
- document store (ex: MongoDB)
- graph database (ex: Neo4J)

---
## An exemple of distributed database: Cassandra

Cassandra is part of the Hadoop ecosystem (it support map reduce task, and cassandra node can be installed on top of hadoop).

Born in Facebook in 2007, Cassandra bases its distribution design on Amazon’s Dynamo and its data model on Google’s Bigtable.

Some of the largest production deployments use Cassandra, including Apple's, with over 75,000 nodes storing over 10 PB of data, Netflix's (2,500 nodes, 420 TB, over 1 trillion requests per day), Chinese search engine Easou's (270 nodes, 300 TB, over 800 million requests per day), and eBay's (over 100 nodes, 250 TB).

.footnote[

![](img/cassandra.png)

]

???

Distributed
Every node in the cluster has the same role. There is no single point of failure. Data is distributed across the cluster (so each node contains different data), but there is no master as every node can service any request.
Supports replication and multi data center replication
Replication strategies are configurable.[16] Cassandra is designed as a distributed system, for deployment of large numbers of nodes across multiple data centers. Key features of Cassandra’s distributed architecture are specifically tailored for multiple-data center deployment, for redundancy, for failover and disaster recovery.
Scalability

Cassandra has innate balancing and rebalancing capabilities in the case of adding or removing nodes.

Cassandra uses **location awareness**, and distinguishes between two levels: the rack, and the data center.

<!-- flexible schema -->

1. The node launches a "snitch" (FR: cafteur), whose task is to find the fastest available node having the requested records
2. The client downloads the records.
3. Simmultaneously, the node asks for the checksums of the same records from the other (slower) nodes that contain the same data. As many checksums are askes as the consistency level specified in the request.

### Gossipping

Once per second, a node will send a message to a random node of the cluster.

If the node does not answer, it marks it locally as down.

<!--Not so clear...-->

.footnote[The term "gossip protocol" dates back from 1987, in an article by Alan Demers.]

Cassandra is typically classified as an AP system, meaning that availability and partition tolerance are generally considered to be more important than consistency in Cassandra,[17] Writes and reads offer a tunable level of consistency, all the way from "writes never fail" to "block for all replicas to be readable", with the quorum level in the middle.

Data is automatically replicated to multiple nodes for fault-tolerance. Replication across multiple data centers is supported. Failed nodes can be replaced with no downtime.

Designed to have read and write throughput both increase linearly as new machines are added, with the aim of no downtime or interruption to applications.

---
## An exemple of distributed database: Cassandra

### Key ideas

Cassandra is **decentralized**: it is a peer-to-peer database, at the opposite spectrum of the a master-slave architecture. Each node functions exactly the same as the other, and no single one is necessary to the overall functionning of the database (no **single point of failure**).

The nodes maintain their knowledge of the network through **gossip**.

Cassandra is a **sparse**, **row-oriented** database, meaning that not all columns should exist for each record. Each row gets a key, that is used to distribute and replicate the records.This allows Cassandra to save space of non-assigned values.

Cassandra fundamental choice is giving up consistency in echange for availability: there is not guarantee that data will be up to date when reading. Cassandra controls the trade-off between consistency and fault-tolerance with 2 parameters:
- **replication factor** is the number of copies the database has to maintain
- **consistency level** is specified at each interaction: it is the number of nodes the databse has to consult and that have to agree for the operation to be considered a success.

---

## An exemple of distributed database: Cassandra

### Typical execution

When a clients wants to write or read, he adresses himself to any of the nodes, while specifying a desired consistency level.

1. The node becomes coordinator for this request.
2. The coordinator forwards the request to all the <!--known?--> replicas hosting the nodes.
3. In case of a write, the replica logs the request in its "commit log" on disk, performs the write in memory, then confirms the transaction success, and returns a result if any.
4. The coordinator waits for as many confirmations as the client has requested with its consistency level before confirming the reading/writing and returns the consolidated response.

---

## An exemple of distributed database: Cassandra

### Replicas

When an write instruction is given, a replica:

1. logs it in a log (called "commit log")
2. pushes the changes in memory ("memtable")
3. if the current memtable reaches a threshold, creates a new memtable
4. when idle, copies the memtable to disk

This ensures a trade-off between availability and fault-tolerance.

---

## An exemple of distributed database: Cassandra

### Properties

#### Fault-tolerance

---
## An exemple of distributed database: Cassandra

### Properties

#### Scalability

---
## An exemple of distributed database: Cassandra

### Properties

#### Consistency

---
# 6. Distributing tasks


---
## Scheduling

1. divide a "job" into "tasks"
2. priorize tasks and divide them into queues
3. get the resources (disk, memory, processing power) to execute these tasks
4. get the job done and return the result

.footnote[There is a lot of blur around the vocabulary here. We will distinguish between a _job_, work asked by a client and to be executed at some point, and a _task_, the miminimal unit of work executable at once. This is not to be confused with the physical counterpart of the effective computation: we there talk about a _process_ for a unit of work that needs resources to be executed (typically disk and memory) and and about a _tread_ for the minimal computing unit sharing the latter resources. This vocabulary, however, is not stabilized: the "Windows Task Manager" manages processes while the "Windows Task Scheduler" manages jobs.]

---

## Specific problems of distributed task management

1. **load balancing** (do not over-use some resources, nor underuse them)
2. target **quality of service** such as:
  - maximizing throughput (the total amount of work completed per time unit);
  - minimizing wait time (time between job submission and first task being executed);
  - minimizing latency or response time (time between job submission and completion);
  - maximizing fairness (equal resource usage allocated to each job, incl. specific clients may have specific access)
3. scalability (clients can be added, so can machines)
4. fault tolerance (not dependant on one single machine)

???

**Sources:**

- https://en.wikipedia.org/wiki/Scheduling_(computing)
- https://stackoverflow.com/questions/3073948/job-task-and-process-whats-the-difference/3073961
- https://arxiv.org/pdf/1901.03270.pdf #TO-READ
- http://www.ijsce.org/wp-content/uploads/papers/v2i3/C0812062312.pdf # TO-READ
- https://en.wikipedia.org/wiki/Parallel_computing #TO-READ
- https://en.wikipedia.org/wiki/Distributed_computing #TO-READ (à partir de Theoretical Fundations)

> Three significant characteristics of distributed systems are: concurrency of components, lack of a global clock, and independent failure of components

> [T]he following defining properties are commonly used as:
> - There are several autonomous computational entities (computers or nodes), each of which has its own local memory.
> - The entities communicate with each other by message passing.
> - The system has to tolerate failures in individual computers.

> A distributed system may [or may not] have a common goal, such as solving a large computational problem; [...] Alternatively, each computer may have its own user with individual needs, and the purpose of the distributed system is to coordinate the use of shared resources or provide communication services to the users.

> The structure of the system (network topology, network latency, number of computers) [may or may not be] known in advance [...] Each computer may [or not] know only one part of the input.

> Parallel computing may be seen as a particular tightly coupled form of distributed computing, and distributed computing may be seen as a loosely coupled form of parallel computing. In parallel computing, all processors may have access to a shared memory to exchange information between processors. In distributed computing, each processor has its own private memory (distributed memory). Information is exchanged by passing messages between the processors.

> the traditional uses of the terms parallel and distributed algorithm that do not quite match the above definitions of parallel and distributed systems 

---
## An exemple of job-scheduler: YARN

YARN is the job-scheduler associated with HDFS.

.center.height400[![](img/hadoop-ecosystem.png)]

.footnote[![](img/hadoop-yarn.png)]

???

**Sources:**

- https://en.wikipedia.org/wiki/Apache_Hadoop
- https://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/YARN.html

---
## An exemple of job-scheduler: YARN

### The architecture

YARN has a slave-master architecture, but not a standard one.

There are two layers:

- the global resource manager, which is in charge of monitoring the resource usage (CPU, memory, storage, network) ; the master (ResourceManager) arbitrates what the resources are being used for, while the slaves (NodeManagers) track the effective usage and report to the master
- the per-application task manager, which is in charge of neogiciating resources with the ResourceManager ; the master (ApplicationMaster) runs on a specific NodeManager and is in charge of checking the tasks progress ; the slaves (resource containers) execute the tasks

???

> YARN allows applications to specify placement constraints in the form of data locality (preference to specific nodes or racks) or (non-overlapping) node labels. (...) For example, it may be beneficial to co-locate the allocations of a job on the same rack (affinity constraints) to reduce network costs, spread allocations across machines (anti-affinity constraints) to minimize resource interference, or allow up to a specific number of allocations in a node group (cardinality constraints) to strike a balance between the two. (...) The applications can specify constraints without requiring knowledge of the underlying topology of the cluster or the other applications deployed. Currently, all constraints are hard, that is, if a constraint for a container cannot be satisfied due to the current cluster condition or conflicting constraints, the container request will remain pending or get rejected.

<!--
## An exemple of job-scheduler: YARN
### Properties

**Fault-tolerance** is guaranteed by ...

**Scalability** is guaranteed by ...



How this specific architecture acheives fault-tolerance, etc.

Hadoop has **location awareness**, i.e. it uses the locations of the nodes relative to each other, in order to allocate the data to the closest nodes.
-->

---
## An exemple of job-scheduler: YARN

### The architecture

.center[![](img/yarn-architecture.gif)

.footnote[source :https://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/YARN.html]

---
## Other job-schedulers

Distributed job-schedulers are sometimes labelled "cluster management software".

- Mesos
- SLURM
- ...

???

- https://en.wikipedia.org/wiki/Slurm_Workload_Manager #TO-READ
- HPC-ABDS #TO-READ

---
## Other job-schedulers

In order to support heterogeneous clusters composed of computers running different OS's, most of job-schedulers use so-called "containers", i.e. extremely light virtual machines. Thus job-scheduling gets quite close to "container orchestration", i.e. the dynamic luanching and management of containers.

- EC2
- Kubernetes
- ...


---
# 7. Parallezing compution on distributed data

---

## Parallezing compution on distributed data

- minimize data transfer
- perform computation at the location of data

---
## Map-reduce principle

As you remember, the map-then-reduce principle is the idea of performing:
- first all embarassingly parallel steps
- then an associative, commutative summarizing rule

-> The second step can be performed as soon as a the first nodes finish completing the first step, and the summaries are usually of smaller size than the original data.

---
## Map-shuffle-reduce principle

On distributed data, the transfer of data becomes increasingly expensive. Thus it is a good idea to minimize tranfer.

Yet, in a typical "group action" (like group count, group average, rolling average...), the same reducer is applied to separately to each group. Group actions are very common in practice.

The **shuffle** step consists in the optimal transfer of data so that all data concerning one group be situated on the same node, while minimizing the overal data transfer, for optimal reduce step.

---
## An exemple of parallelized computation framework: Spark

Spark is a project supported by the Apache fundation.

It is written mostly in Java (for portability), but exposes APIs in Java, Scala, Python and R, even though not all functionnalities are available in all languages.

Specificities:

- handling of failures
- lazy evaluation

.footnote[![](img/spark-logo.png)]

???

**Sources:**
- https://stackoverflow.com/questions/29011574/how-does-spark-partitioning-work-on-files-in-hdfs
- https://stackoverflow.com/questions/38284210/how-does-a-spark-task-access-hdfs
- https://sujithjay.com/spark/with-yarn
- Hig Perfromence Spark, O'Reilly, (ch.2)

---
## An exemple of parallelized computation framework: Spark

### Key ideas

The usual way to go is to run a Spark job:

1. You create a Spark _application_, in charge of maintaining communication with the the cluster. The application asks for resources to be attributed, typically on each node of the cluster.
2. The cluster's job-scheduler reserves resources for the Spark application, until termination.
3. The Spark application launches Java virtual machines (JVMs) on the nodes, and ships the code necessary for the execution of the different tasks.
4. Each job (one specific computation) is decomposed as elementary tasks along a _directed acyclic graph_ (DAG)
5. Parallelisable tasks are computed the closest possible to the data blocks
6. Intermediary results are shufled, then summarized (reduce step)
7. Result is returned to the application

---
## An exemple of parallelized computation framework: Spark

### Key ideas

This means:

- two Spark applications do not share virtual machine on the same computer, thus jobs submitted via different applications can not share data (besides writing to external support such as disk / database)
- when running interactively, there is no "termination" or "completion" ; the Spark session will usually hold on to the resources, preventing other users / other jobs to be run on these nodes in the meanwhile

---
## An exemple of parallelized computation framework: Spark

### Architecture

Spark has a master-slave architecture: the master is called the _driver_, and the slaves the _executors_. Whereas _executors_ are typically situated on the same nodes as HDFS and YARN slaves, the _driver_ is typically a client outside the cluster.

---
## An exemple of parallelized computation framework: Spark

### Partitions

Data is typically first loaded by Spark, in memory or onto disk, creating an object distributed accross all executors called a _partition_. If the executors live on the same nodes as some of the data blocks they are willing to read, then data is read locally. Otherwise, Spark will ship data blocks from the closest available DataNode.

Every _split_ of a _partition_ has a "prefered location", that depends ultimately on where it was stored in the distirbuted file system / database. Spark will try to move each split as little as possible.

---
## An exemple of parallelized computation framework: Spark

### Lazy evaluation & execution plans

Spark only records the succession of _transformations_ you want to perform, and only performs the computation on specific _actions_, such as collecting a result bak to the _driver_ or writing in the file system.

At this moment, the _driver_ builds a _directed acyclic graph_ (DAG), of all the intermediary steps needed to obtain this result, going backward from the result. This series of steps is called the _execution plan_.

The execution plan voids unneeded computation that may have wrongly added to the transformation chain. Typically, all filter actions will be performed first, so that you don't compute things that you do not use later. And all map steps are grouped together.

The execution of one DAG is a _job_ for the cluster's task-scheduler. One Spark application can thus launch several jobs, as many as actions.

---
## An exemple of parallelized computation framework: Spark

### Lazy evaluation & execution plans

This strategy has several advantages:
- filter operations actually stop any further computation
- all map steps can be computed in a single pass on the data
- any intermediary result can be recomputed in the case of a node failure

---
## An exemple of parallelized computation framework: Spark

### Persistence

- Does not store data on the long term, only for the life of the application
- Typical usage is to return results to the driver / to write the output on a network file
- Does not care about storage failure (it prefers re-computing from the DAG instead)
- Normal behaviour:
    - Stores all computed intermediary results as distinct values in memory, until memory limit is exceeded. This allows recycling earlier computations for new execution plans.
    - When memory limit is exceeded, new results replace the older.
    - If a single step exceeds the memory limits, data is stored to disk.
- It is possible to control the caching behaviour and forcing some intermediary results to persist and to chose where it has to persist.

---
## An exemple of parallelized computation framework: Spark

### Client mode architecture

.center.height400[![](img/spark client mode.jpg)]
<!-- --- -->
<!-- ## Other frameworks -->

<!-- --- -->
<!-- # 8. Statistical applications -->


---
# 9. Conclusions and perspectives

---
## What to remember from today?

- When use a file system or a database
- The challeng of distribution
- Master-slave architecture
- The main features Hadoop works
- The main features Cassandra works
- What is Yarn
- The main features of Spark

---
## What to remember from the course?

Is big data a problem for you?
Is it worth it?

Key insights:
- use sampling
- maybe you can afford waiting
- beware of the cost
- low level is faster
- think befor you do

<!-- --- -->
<!-- ## Now you understand this : -->





???

--> quotes explaining what this or this software is


<!-- fast languages must be moved to the first course:
Fast languages (Go, Julia, JavaScript) -->


<!--

THINGS TO ADD SOMEWHERE:

From Wikipedia:

"The master node consists of a Job Tracker, Task Tracker, NameNode, and DataNode."
"A slave or worker node acts as both a DataNode and TaskTracker, though it is possible to have data-only and compute-only worker nodes."
Communication between the nodes happens in SSH.
"In a larger cluster, HDFS nodes are managed through a dedicated NameNode server to host the file system index, and a secondary NameNode that can generate snapshots of the namenode's memory structures, thereby preventing file-system corruption and loss of data."


From https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html

HDFS relaxes a few POSIX requirements to enable streaming access to file system data.
HDFS applications need a write-once-read-many access model for files. A file once created, written, and closed need not be changed except for appends and truncates. Appending the content to the end of the files is supported but cannot be updated at arbitrary point. This assumption simplifies data coherency issues and enables high throughput data access. A MapReduce application or a web crawler application fits perfectly with this model.
A computation requested by an application is much more efficient if it is executed near the data it operates on. This is especially true when the size of the data set is huge. This minimizes network congestion and increases the overall throughput of the system. The assumption is that it is often better to migrate the computation closer to where the data is located rather than moving the data to where the application is running. HDFS provides interfaces for applications to move themselves closer to where the data is located.
To minimize global bandwidth consumption and read latency, HDFS tries to satisfy a read request from a replica that is closest to the reader.

The HDFS architecture is compatible with data rebalancing schemes. A scheme might automatically move data from one DataNode to another if the free space on a DataNode falls below a certain threshold. In the event of a sudden high demand for a particular file, a scheme might dynamically create additional replicas and rebalance other data in the cluster. These types of data rebalancing schemes are not yet implemented.

-->

<!--
- Consistency & Availability:
    Such database 

- Availability & Fault-tolerance:

- Fault-Tolerance & Consistency:
    If you shard your data to distribute it across several nodes.

> « We encounter scalability problems when our relational applications become successful and usage goes up. Joins are inherent in any relatively normalized relational database of even modest size, and joins can be slow. The way that databases gain consistency is typically through the use of transactions, which require locking some portion of the database so it’s not available to other clients. This can become untenable under very heavy loads, as the locks mean that competing users start queuing up, waiting for their turn to read or write the data. »

Extrait de: Jeff Carpenter. « Cassandra: The Definitive Guide: Distributed Data at Web Scale. »

> « We encounter scalability problems when our relational applications become successful and usage goes up. Joins are inherent in any relatively normalized relational database of even modest size, and joins can be slow. The way that databases gain consistency is typically through the use of transactions, which require locking some portion of the database so it’s not available to other clients. This can become untenable under very heavy loads, as the locks mean that competing users start queuing up, waiting for their turn to read or write the data. »

Extrait de: Jeff Carpenter. « Cassandra: The Definitive Guide: Distributed Data at Web Scale. » Apple Books. 

> « We encounter scalability problems when our relational applications become successful and usage goes up. Joins are inherent in any relatively normalized relational database of even modest size, and joins can be slow. The way that databases gain consistency is typically through the use of transactions, which require locking some portion of the database so it’s not available to other clients. This can become untenable under very heavy loads, as the locks mean that competing users start queuing up, waiting for their turn to read or write the data. »

« We encounter scalability problems when our relational applications become successful and usage goes up. Joins are inherent in any relatively normalized relational database of even modest size, and joins can be slow. The way that databases gain consistency is typically through the use of transactions, which require locking some portion of the database so it’s not available to other clients. This can become untenable under very heavy loads, as the locks mean that competing users start queuing up, waiting for their turn to read or write the data. »

Extrait de: Jeff Carpenter. « Cassandra: The Definitive Guide: Distributed Data at Web Scale. » Apple Books. 

« The problems that 2PC introduces for application developers include loss of availability and higher latency during partial failures. Neither of these is desirable. So once you’ve had the good fortune of being successful enough to necessitate scaling your database past a single machine, you now have to figure out how to handle transactions across multiple machines and still make the ACID properties apply. Whether you have 10 or 100 or 1,000 database machines, atomicity is still required in transactions as if you were working on a single node. But it’s now a much, much bigger pill to swallow. »

Extrait de: Jeff Carpenter. « Cassandra: The Definitive Guide: Distributed Data at Web Scale. » Apple Books. 

Extrait de: Jeff Carpenter. « Cassandra: The Definitive Guide: Distributed Data at Web Scale. » Apple Books. 

???

**Source:**

- https://en.wikipedia.org/wiki/Denormalization
- Cassandra, the definitive guide
-->