Distributed File Systems - Mabel

Question 1

Where does data come from?

Answer 1

• ˆ Raw data: sensors, measurements, events, logs (e.g. CERN sensor measurements)
• ˆ Derived data: aggregated data, intermediate data (e.g. computational results on sensor measurements)

Question 2

How would Billions of TB files be stored?
vs
How would Millions of PB files be stored?

Answer 2

Object Storage - Key Value Model (like in S3) + Object Storage (lots of files but single files aren’t very large)
vs
File Storage - File System (brings back the file hierarchy natively) + Block Storage (supports block-based storage natively)

Question 3

Fault Tolerance in Clusters: How likely are Local Disks and Clusters likely to fail?

Answer 3

• Local disk: the disk might fail, we can just keep a backup in case.
• Cluster with 100s to 10’000s machines: nodes will fail (P [at least one node fails ] = 1 − (1 − p)^n, with n: #nodes, p: failure probability of a single node). We thus need a stronger property for clusters, we can’t
  restore from backups every single day.

Question 4

How to achieve fault tolerance and robustness on clusters?

Answer 4

• Fault tolerance (system keeps working under faults)
• Automatic Recovery (can’t recover manually with lots of disks)
• Error detection (know about failed disks)
• Monitoring (keep an overview over what’s working)

Question 5

What kind of ways do we want to read/write large files in DFS?

Answer 5

• Efficiently scan a large file (in its entirety) – for data analysis
• Append efficiently new data at the end of an existing large file – particularly for logging and sensors

Furthermore, this must be supported even with hundreds of con- current users reading and writing to and from the same file system.

Random Access is very difficult in clusters

Question 6

What performance requirements do we have for DFS?

Answer 6

The bottleneck must be Throughput (how fast we can read/write), NOT LATENCY

This is more consistent with the full-scan pattern we use

Question 7

How do we solve the capacity-throughput discrepancy?

Answer 7

The solution for the capacity-throughput discrepancy is to parallelize

Question 8

How do we solve the throughput-latency discrepancy?

Answer 8

The solution for the throughput- latency discrepancy is to do batch-processing

Question 9

What 3 components is Hadoop primarily composed of?

Answer 9

• Distributed File System (HDFS) (inspired by Google’s GFS)
• MapReduce (inspired by Google’s MapReduce)
• Wide column store (HBase) (inspired by Google’s BigTable)

Question 10

Describe the model for distributed file systems?

Seperating the logical model and physical storage

Answer 10

File system (logical model): In distributed file systems, we have a file hierarchy (unlike the key-value model in object storage which is flat).
Block storage (physical storage): In distributed file systems, we have block storage (unlike object storage (S3) where we have a blackbox model).

Terminology:** HDFS: Block**, GFS: Chunk. We thus have a hierarchy of files where each file is associated with a chain of blocks.

Question 11

Why do we used blocks in DFS?

Answer 11

1. Files can be bigger than individual disks (PBs)
2. Simple level of abstraction and blocks are easy to distribute over multiple nodes.

Question 12

What is the size of a block in DFS?

Answer 12

64-128MB

Question 13

Why is a DFS Block 64-128MB?

Answer 13

Due to the throughput-latency discrepancy, we want larger blocks (for small blocks, the latency would outweigh the transfer time). Further, large blocks lead to less blocks being read per file. The block size in distributed file systems is 64MB - 128MB. This is a good compromise - not too many blocks for big files, but also small enough to have several blocks on one machine.

Question 14

What architecture does HDFS use?

Answer 14

HDFS uses a centralised architecture: the namenode (has the names of the files) is the master and the datanodes (have the actual data) are the workers.

Question 15

How is data actually stored in HDFS?

Answer 15

The file is divided into 128MB chunks. The chunks are then stored in datanodes. Each chunk is replicated 3 times (the # of replicas can be specified).

Question 16

What three activities is the NameNode responsible for?

Answer 16

1. File namespace (+Access Control): keep track of hierarchy of files. This is actually rather small (hierarchy doesn’t contain the actual file data, just the hierarchy).
2. File to block mapping: every file is associated with a list of blocks. The namenode keeps track of the mapping file
   3.** Block location**: for every block, the namenode needs to know on which 3 (default) datanodes the block is stored.

Question 17

What are the DataNodes responsible for?

Answer 17

Blocks of data are stored on these local disks. Datanodes are responsible for failure detection. Each datanode has its own local view over its disks - proximity to hardware facilitates disk failure detection. Each block has a block ID (64bit). It is also possible to access blocks at a subblock granularity to request parts of a block.

Question 18

How do DataNodes and NameNodes interact?

Answer 18

The datanode always initiates the connection.
* Registration
* Heartbeat: datanode tells namenode every 3s that it’s still alive
* Block Report: every 6h, datanode sends full list of blocks (not the contents of the blocks) to the namenode
* Block Received

The datanode protocol handles datanode-namenode communication.

Question 19

What is the Client Protocol?

Answer 19

Clients sends metadata operations (e.g. create directory, delete directory, write file, append to file, read file, delete file) and the namenode responds with the datanode location and the block IDs

Question 20

Can DataNodes connect to one another?

Answer 20

Yes! DataNodes are also capable of communicating with each other by forming replication pipelines. A pipeline happens whenever a new HDFS file is created. The client does not send a copy of the block to all the destination DataNodes, but only to the first one. This first DataNode is then responsible for creating the pipeline and propagating the block to its counterparts.

Question 21

How are files read from HDFS?

Answer 21

1. Client asks namenode for file
2. Namenode sends block location (multiple datanodes for each block, sorted by distance) to client
3. Client reads data from datanode via input stream

Question 22

How is a file written in HDFS?

Answer 22

1. Client sends create to namenode
2. Namenode sends datanodes (all replicas) for first block
3. Client contacts one datanode and tells it which others to organise replication pipeline with
4. Client sends the data over to first node
5. Datanode sends Ack to client which informs namenode.
6. Namenode sends client datanodes for second block (writing happens block by block)
7. Repeat
8. Client asks Name node to close/release lock
9. Datanodes check with namenode for minimal replication (datanode protocol)
10. Namenode sends Ack to client
11. Namenode can tell datanodes to replicate further asynchronously

All acks must be returned to move on to next block

Question 23

How does replication physically happen in HDFS?

A cluster contains multiple racks and a rack contains multiple nodes

Answer 23

• Replica 1: same node as client (or random) - rack A- because it’s very efficient
• Replica 2: a node in a different rack B (put in a different rack from 1st replica - racks can fail)
• Replica 3: in same rack B but on a different node
• Replica 4 and beyond: random, but if possible: – at most one replica per node – at most two replicas per rack

Question 24

How do we define distance in a cluster?

Answer 24

Like a tree!

Two nodes in the same rack have a distance of 2 (one edge from the first node to the rack, and one from the rack to the other node).

Two nodes in different racks have a distance of 4

Question 25

Why don't we place replicas 1 & 2 on the same rack? (client rack)

Answer 25

Because then 2/3 of replicas of all blocks written by the client will be on the same rack (the client stays on the same node

Question 26

What is the single point of failure of HDFS?

Answer 26

**Namenode** If the metadata stored on it is lost, then all the data on the cluster is lost, because it is not possible to reassemble the blocks into files any more.

Question 27

How do we protect from NameNode failure?

Answer 27

Backups! The** file namespace **containing the directory and file hierarchy as well as the **mapping from files to block IDs** is backed up to a 'snapshot'. Updates to the file system arriving after the snapshot has been made are instead stored in a journal, called edit log.

Question 28

How do we use the backup to recover from a NameNode failure?

Answer 28

Restore the initial hierarchy/ block mapping from 'snapshot' and then ’play’ the edit log and apply changes to the file systems (restore last logged version) Receive the block locations from the block reports, which are periodically sent by the datanode (can also be manually requested)

Question 29

How do we reduce the time taken to restart the system after a crash? | Start up with basic HDFS > 30 mins

Answer 29

1. Edit log is periodically merged back into a new, larger snapshot and reset to an empty edit log. This is called a checkpoint and is performed by a "phantom" NameNode with the same structure 2. Configure a phantom NameNode to be able to instantly take over from the NameNode in case of a crash. This considerably shortens the amount of time to recover (high availability) 3. Federated NameNodes are non-overlapping NameNodes responsible for sections of data thus are faster

Question 30

How do you move log data to HDFS?

Answer 30

**Apache Flume**: Collects, aggregates, moves log data (into HDFS)

Question 31

How do you import from a relational database to HDFS?