�ݺ�ߣ

Hadoop in action
Gabriele Lombari - Alessandra Zullo
Sistemi Operativi Avanzati - Giuseppe Cattaneo
April 21, 2016
Universit`a degli Studi di Salerno

Table of contents
1. Esempio: WordCount
2. Apache Hadoop & MapReduce
3. Hadoop Distributed File System
4. Hadoop v2.x
5. WordCount Distribuito
6. Prestazioni
7. Hadoop in action!
8. Riferimenti
2

Esempio: WordCount (1)
Immaginiamo di avere un grande DataSet di parole
{web, weed, green, sun, moon, land, part, web, green ...}
di cui volgiamo contare il numero di occorrenze.
4

Esempio: WordCount - Sequential Code
1
2 p u b l i c c l a s s SequentialWordCount {
3
4 p u b l i c s t a t i c void main ( S t r i n g [ ] args ) throws Exception {
5
6 F i l e I n p u t S t r e a m fstream = new F i l e I n p u t S t r e a m ( ”BIG FILE 8GB . t x t ” ) ;
7 DataInputStream i n = new DataInputStream ( fstream ) ;
8 BufferedReader br = new BufferedReader ( new InputStreamReader ( i n ) ) ;
9
10 HashMap<String , I n t e g e r> map = new HashMap<>();
11
12 S t r i n g word ;
13 w h i l e (( word = br . re adL in e () ) != n u l l ){
14
15 S t r i n g T o k e n i z e r t o k e n i z e r = new S t r i n g T o k e n i z e r ( word , ” tnrf . , ; : ’ ” ” ) ;
16
17 w h i l e ( t o k e n i z e r . hasMoreTokens () ) {
18 S t r i n g word now = t o k e n i z e r . nextToken () ;
19 I n t e g e r count = map . get ( word now ) ;
20
21 map . put ( word now , count == n u l l ? 1 : ++count ) ;
22 }
23 }
24
25 f o r ( Entry<String , I n t e g e r> e n t r y : map . e n t r y S e t () ) System . out . p r i n t l n ( e n t r y . getKey () + ” : ” +
e n t r y . getValue () ) ;
26 }
27
28 }
5

Esempio: WordCount - MapReduce execution
Un esempio di come il problema WordCount viene risolto in
un’esecuzione su ambiente MapReduce.
6

Un po’ di storia
La genesi di Apache Hadoop ha avuto inizio quando Google presenta,
nell’Ottobre del 2003, The Google File System e, poi nel Dicembre
2004, MapReduce: Simpliﬁed Data Processing on Large Clusters.
Lo sviluppo del framework inizia con il progetto Nutch, divenuto presto a
far parte del progetto Apache Lucene.
Nel 2005 Yahoo! entra a far parte del progetto Nutch decretando cos`ı
l’inizio di Hadoop.
8

Un po’ di storia
Il progetto fu supervisionato da Doug Cutting e Michael J. Cafarella
Il nome Hadoop fu coniato dal nome del pupazzo del ﬁglio di Cutting.
9

Introduzione
Apache Hadoop `e un framework per l’elaborazione distribuita di grandi
insiemi di dati(Big Data) che usa il modello di programmazione
MapReduce.
10

MapReduce (1)
Il modello computazionale proposto da Google nel 2004 prevedeva che gli
utenti dovessero comporre la loro logica di elaborazione basandosi su due
soli step:
11

MapReduce (1)
soli step:
• Map, il quale prende in input una coppia < K, V > e restituisce una
lista di coppie < K , V >, intermedie;
11

MapReduce (1)
soli step:
• Reduce, il quale prende in input una lista di coppie < K , V >,
aventi la stessa K, e restituisce in output una coppia < K , V >.
11

MapReduce (1)
soli step:
• Reduce, il quale prende in input una lista di coppie < K , V >,
aventi la stessa K, e restituisce in output una coppia < K , V >.
Ciò permette una grande parallelizzazione in grado di sfruttare al meglio
le nostre CPU, eliminando qualsiasi difficoltà al programmatore.
11

MapReduce (2)
Al programmatore, quindi, verr`a lasciato solo il compito di scrivere il
codice per le operazioni di Map e Reduce, sar`a compito del framework
rendere parallela la computazione in maniera del tutto trasparente.
12

MapReduce (2)
In particolare il framework si occupa di:
12

MapReduce (2)
• Scheduling;
12

MapReduce (2)
• Scheduling;
• Data/code co-location;
12

MapReduce (2)
• Scheduling;
• Synchronization;
12

MapReduce (2)
• Scheduling;
• Synchronization;
• Error and fault handling.
12

Apache Hadoop
Hadoop è formato principalmente da 4 macro-progetti:
• Hadoop Common: librerie di base per la corretta esecuzione del
framework;
• Hadoop Distributed File System (HDFS): file system distribuito
che fornisce un accesso veloce ai dati;
• Hadoop Yet Another Resource Negotiator (YARN): sistema per
la pianificazione dei processi e la gestione delle risorse del cluster.
• Hadoop MapReduce: sistema per l’elaborazione parallela dei
BigData;
13

Hadoop Distributed File System

Hadoop Distributed File System - HDFS (1)
HDFS, acronimo di Hadoop Distributed File System, è il file system
distribuito di Hadoop.
File system distribuito (DFS): particolare file system che permette la
memorizzazione di file e risorse in dispositivi di archiviazione
distribuiti in rete, resi disponibili attraverso un meccanismo client-server
tra dispositivi remoti.
15

HDFS è un file system strutturato a blocchi;i file sono suddivisi in
blocchi di dimensioni fisse, non necessariamente memorizzati sulla stessa
macchina.
16

macchina.
Le dimensioni sono di 64MB per Hadoop v1.x e 128MB per Hadoop v2.x.
16

macchina.
HDFS non pu`o essere montato direttamente sul SO esistente, per tale
motivo bisogna spostare i ﬁle prima e dopo l’esecuzione di un job.
16

macchina.
HDFS non pu`o essere montato direttamente sul SO esistente, per tale
motivo bisogna spostare i ﬁle prima e dopo l’esecuzione di un job.
HDFS prevede un modulo che permette alle applicazioni di spostarsi
vicino ai dati in modo da minimizzare la congestione della rete e
incrementare il throughput del sistema.
16

HDFS - Architettura (1)
HDFS ha un’architettura master/slaves.
Master e slave sono indicati con il nome dei daemons che forniscono il
rispettivo servizio; con Namenode indichiamo il master e con Datanode
indichiamo gli slaves.
17

HDFS - Architettura(2)
In dettaglio:
• master (uno): tiene traccia del file namespace (metadati che
contengono informazioni circa il file, la struttura delle directory di
tutti i file presenti sull’HDFS,la posizione dei blocchi e i permessi di
accesso);
• slaves (molti): effettua il reale calcolo computazionale; memorizza e
recupera i blocchi quando richiesto (dal client o dal master), e
comunica periodicamente al master la lista dei blocchi da esso
memorizzati. .
18

HDFS - Architettura(3)
È importante sottolineare che un failure del Namenode causerebbe un
fault di tutto il sistema.
Tale problema è stato risolto in Hadoop v2.x con l’aggiunta del modulo
Hadoop HA in cui vi è una coppia di NameNode (active e standby);
nel caso in cui si verifica una failure sul NameNode active Hadoop HA
si occupa, in maniera trasparente, di attivare il NameNode stanby come
NameNode principale.
I DataNode comunicano tra di loro per bilanciare i dati,spostare delle
copie o per mantenere delle repliche dei dati.
19

HDFS - Data Replication
L’HDFS è progettato per memorizzare e gestire molti grandi file.
Ogni file è scomposto in blocchi, ciascuno dei quali è memorizzato in un
DataNode; per fault tolerance i blocchi sono replicati.
La replica su macchine fisiche differenti incrementa la località dei dati e
migliora l’efficienza dell’esecuzione MapReduce.
20

HDFS - Scrittura di un file (1)
Quando un client chiede di scrivere un file sull’HDFS esso viene
memorizzato in una cache temporanea (in maniera trasparente), se tale
file supera una certa grandezza allora viene contattato il NameNode, che
alloca un blocco di dati e lo inserisce nella gerarchia del file system.
Il NameNode risponde al client restituendogli l’identità (id) del
DataNode e il datablock di destinazione su cui viene memorizzato il
contenuto del file temporaneo; infine viene svuotata la cache.
21

HDFS - Scrittura di un ﬁle (2)
22

HDFS - Replication Pipelining
Quando si scrive un ﬁle sull’HDFS il NameNode decide il DataNode su
cui memorizzare il blocco (prima replica), tale DataNode invia una copia
al prossimo nodo (seconda replica) e cos`ı via.
In tal modo ogni DataNode `e pipeline al nodo successivo.
23

HDFS - Lettura di un file
Quando un client vuole leggere un file, contatta il NameNode
richiedendo lo specifico file; il NameNode restituisce le informazioni sui
datablocks e la posizione dei DataNodes.
Il client, grazie alle informazioni passate dal NameNode, unisce i singoli
blocchi per ricostruire il file.
24

HDFS - Many small file
Uno small file è un file di dimensioni molto più piccole rispetto alle
dimensioni di un blocco dell’HDFS, questo significa che ogni small file è
inviato ad un map task.
25

Il NameNode salva in memoria le informazioni per ogni ﬁle; avere molti
small ﬁle comporta un overhead per esso.
25

Il NameNode salva in memoria le informazioni per ogni file; avere molti
small file comporta un overhead per esso.
SOLUZIONE: Si adopera un container per raggruppare tutti i file in un
unico file più grande.
25

Hadoop v2.x (1)
Hadoop v2.x è composto da miglioramenti applicati alle release stabili di
Hadoop v1.x.
II principali cambiamenti riguardano:
• YARN ( Yet Another Resource Negotiator ), un nuovo sistema di
runtime MapReduce (ma non il SOLO!) ed un gestore di risorse
distribuite.
• HDFS Federation, partiziona l’HDFS namespace in più NameNode
per poter usare un cluster con molti più file.
• NameNode High-availability, risolve eventuali failure attivando un
NameNode che è in standby per failover.
27

Hadoop v2.x - Componenti principali (1)
Le principali componenti sono:
• Global Resource Manager (RM) che gestisce l’assegnamento
globale delle risorse di calcolo alle applicazioni;
• Application Master (AM) gestisce l’applicazione, lo scheduling e
l’esecuzione dei task; ogni applicazione ha il suo AM.
• Container che rappresenta una risorsa allocata nel cluster ed è
sempre assegnato ad un singolo job.
• NodeManager, (uno per ogni nodo) si occupa di gestire la
computazione sul singolo nodo, mantiene informato il
ResourceManager sulle sue attività e inoltre è responsabile per i
Container e per il monitoraggio delle risorse usate (cpu, memoria,
disco e rete) da quel nodo.
Per applicazione si intende o un singolo job o un DAG dei job.
28

Hadoop v2.x - Componenti principali (2)
29

Hadoop MapReduce
Come gi`a detto in precedenza, Hadoop `e un framework per l’elaborazione
distribuita di BigData tramite il modello di programmazione MapReduce.
30

Hadoop MapReduce - Splitter (1)
Lo Splitter `e quella componente che ha il compito di prendere l’intero
input e di dividerlo in InputSplits, che verranno poi inviati ai vari map
task.
31

Lo Splitter `e quella componente che ha il compito di prendere l’intero
input e di dividerlo in InputSplits, che verranno poi inviati ai vari map
task.
La mappatura < K, V > degli InputSplits viene deﬁnita dalla classe
InputFormats implementata.
31

Hadoop mette a disposizione diverse implementazioni di InputFormat, tra
cui:
Classe Descrizione
TextInputFormat
Ogni linea è un record
K1: Offset
V1: Testo
KeyValueTextInputFormat
K1: Testo prima di un delimitatore
V1: Testo dopo il delimitatore
SequenceFileInputFormat<K,V>
Utilizzato per record binari
K1: Input di taglia K
V1: Input di taglia V
NLineInputFormat
K1: Offset
V1: Testo 32

Qui è possibile prendere visione della gerarchia delle classi degli
InputFormat messi a disposizione dal framework:
È possibile, ovviamente, definire un proprio InputFormat da utilizzare
durante la computazione implementando l’interfaccia InputFormat.
33

Attenzione!
Non bisgona confondere l’InputSplit con il blocco utilizzato dall’HDFS. I due
”blocchi” non necessariamente corrispondono.
Possono infatti veriﬁcarsi dei casi in cui l’InputSplit utilizza dati presenti in pi`u
blocchi, causando un leggero overhead del sistema.
34

HadoopMapReduce - Partitioners and Combiners (1)
Quella vista ﬁno ad ora `e la parte funzionale del modello presentato da
Google, eccezion fatta per lo Splitter.
36

´E necessario introdurre due elementi addizionali, propri di Hadoop, senza
i quali tutto il framework potrebbe non funzionare o risultare comunque
poco eﬃcente:
36

poco eﬃcente:
• Partitioners;
36

poco eﬃcente:
• Partitioners;
• Combiners.
36

• Partitioners: responsabili di dividere le coppie < K, V > intermedie
generati dai MapTasks ai vari Reducers per la funzione di Reduce.
Per dividere in maniera equa i compiti ai Reducers i partitioners
usano una funzione di hash sulla chiave:
reducer = hash(K) mod n.
37

• Partitioners: responsabili di dividere le coppie < K, V > intermedie
generati dai MapTasks ai vari Reducers per la funzione di Reduce.
Per dividere in maniera equa i compiti ai Reducers i partitioners
usano una funzione di hash sulla chiave:
reducer = hash(K) mod n.
• Combiners: un’ottimizzazione di Hadoop rispetto a MapReduce in
quanto permettono di aggregare LOCALMENTE prima della fasi di
shuﬄe e sorting.
37

38

Hadoop MapReduce (4)
Possiamo riassumere un’esecuzione di MapReduce in:
39

1. Lettura dei dati;
39

2. InputSplitting;
39

2. InputSplitting;
3. Mapping;
39

2. InputSplitting;
3. Mapping;
4. Combining;
39

2. InputSplitting;
3. Mapping;
4. Combining;
5. Partitioning, Shuﬄing and Sorting;
39

2. InputSplitting;
3. Mapping;
4. Combining;
5. Partitioning, Shuﬄing and Sorting;
6. Reducing.
39

Esempio: WordCount - Mapper SourceCode
1 p u b l i c s t a t i c c l a s s TokenizerMapper
2 extends Mapper<Object , Text , Text , I n t W r i t a b l e>{
3 p r i v a t e f i n a l s t a t i c I n t W r i t a b l e one = new I n t W r i t a b l e (1) ;
4 p r i v a t e Text word = new Text () ;
5 p u b l i c void map( Object key , Text value , Context context
6 ) throws IOException , I n t e r r u p t e d E x c e p t i o n {
7 S t r i n g T o k e n i z e r i t r = new S t r i n g T o k e n i z e r ( v a l u e . t o S t r i n g () ) ;
8 w h i l e ( i t r . hasMoreTokens () ) {
9 word . s e t ( i t r . nextToken () ) ;
10 context . w r i t e ( word , one ) ;
11 }
12 }
13 }
La classe Mapper.
41

Esempio: WordCount - Reducer SourceCode
1 p u b l i c s t a t i c c l a s s IntSumReducer
2 extends Reducer<Text , I n t W r i t a b l e , Text , I n t W r i t a b l e> {
3 p r i v a t e I n t W r i t a b l e r e s u l t = new I n t W r i t a b l e () ;
4
5 p u b l i c void reduce ( Text key , I t e r a b l e<I n t W r i t a b l e> values , Context context )
6 throws IOException , I n t e r r u p t e d E x c e p t i o n {
7 i n t sum = 0;
8 f o r ( I n t W r i t a b l e v a l : v a l u e s ) {
9 sum += v a l . get () ;
10 }
11 r e s u l t . s e t (sum) ;
12 context . w r i t e ( key , r e s u l t ) ;
13 }
La classe Reducer.
42

Esempio: WordCount - Main SourceCode
1 p u b l i c s t a t i c void main ( S t r i n g [ ] args ) throws Exception {
2 C o n f i g u r a t i o n conf = new C o n f i g u r a t i o n () ;
3 Job job = Job . g e t I n s t a n c e ( conf , ”word count ” ) ;
4 job . s e t J a r B y C l a s s ( WordCount . c l a s s ) ;
5 job . setMapperClass ( TokenizerMapper . c l a s s ) ;
6 job . setCombinerClass ( IntSumReducer . c l a s s ) ;
7 job . s e t R e d u c e r C l a s s ( IntSumReducer . c l a s s ) ;
8 job . setOutputKeyClass ( Text . c l a s s ) ;
9 job . setOutputValueClass ( I n t W r i t a b l e . c l a s s ) ;
10 F il e In p u tF o r ma t . addInputPath ( job , new Path ( args [ 0 ] ) ) ;
11 FileOutputFormat . setOutputPath ( job , new Path ( args [ 1 ] ) ) ;
12 System . e x i t ( job . waitForCompletion ( t r u e ) ? 0 : 1) ;
13 }
14
Main method.
43

Prestazioni
Per misurare le prestazioni usiamo:
• Speedup che misura l’incremento delle prestazioni all’aumentare dei
processori:
S(n) = Timeseq/TimeHad(n)
dove:
• Timeseq è il tempo di esecuzione di un’implementazione sequenziale
• TimeHad(n) è il tempo di esecuzione di un’implementazione
distribuita che utilizza n slaves.
• Efficienza che esprime la capacità di sfruttare la potenza di calcolo:
E(n) = S(n)/n
dove:
• S(n) è lo speedup
• n è il numero di slaves(core) che eseguono i task Hadoop.
45

Comandi utili
• start-dfs.sh //Avvio di HDFS (NameNode, DataNode)
• stop-dfs.sh //Arresto HDFS
• start-yarn.sh //Avvio del ResourceManager
• stop-yarn.sh //Arrwto del ResourceManager
• mr-jobhistory-daemon.sh start historyserver //Avvio de
JobHistoryServer
• mr-jobhistory-daemon.sh stop historyserver //Arresto del
JobHistoryServer
• hdfs dfs -mkdir -p . //Creazione della cartella radice nell’HDFS
• hdfs dfs -rm <-r> Rimuove ﬁle <cartelle>dall’HDFS
• hdfs dfs -ls -l Visualizzazione del contenuto dell’HDFS
47

Riferimenti
Apache Hadoop
http : //hadoop.apache.org/
https : //hadoop.apache.org/docs/r1.2.1/mapredtutorial.html
http : //hadoop.apache.org/docs/r2.5.1/index.html
MapReduce: Simplifed Data Processing on Large Clusters Dean, J.
and Ghemawat, S. Communication of ACM 51, 1 (Jan. 2008), 107-113
The Hadoop Distributed File System Shvachko, Konstantin et al.
2010 IEEE 26th Symposium on Mass Storage Systems and Technologies
(MSST)
Pro Hadoop Jason Venner - Copyright c 2009 by Jason Venner -
ISBN:978-1-4302-1942-2
50

Riferimenti
Data-Intensive Text Processing with MapReduce Jimmy Lin and
Chris Dyer - University of Maryland, College Park - Manuscript prepared
April 11, 2010
�ݺ�ߣ Ph.D. Roscigno Gianluca Dipartimento di Informatica -
Universit`a degli Studi di Salerno, I-84084, Fisciano (SA), Italy
giroscigno@unisa.it
51

�ݺ�ߣ

Hadoop in action!

More Related Content

Hadoop in action!