UNIVERSIT `
A DI PISA
Dipartimento di Ingegneria dell’Informazione
Corso di Laurea Magistrale in Computer Engineering
Tesi di Laurea in
Progettazione e realizzazione del batch
layer di un’infrastruttura Big Data
basata su architettura Lambda
Relatori:
Prof. Mario G.C.A. Cimino Prof.ssa Gigliola Vaglini Dott. Romeo Zitarosa
Candidato: Antonino D’Alberti Anno Accademico 2018/2019
Indice
1 Introduzione 5
1.1 Elaborazione in batch e Data lake . . . 5
1.2 Industria 4.0: sfide future . . . 6
2 Infrastruttura Big data 8 2.1 Architettura Lambda . . . 8
2.2 Aggregate-Oriented data model . . . 10
2.3 MapReduce . . . 10
2.3.1 Modello di programmazione . . . 11
2.3.2 Esecuzione . . . 11
2.4 Complessit`a nelle infrastrutture big data . . . 11
3 Tecnologie Utilizzate 13 3.1 Apache Spark . . . 13 3.1.1 Un unico stack . . . 14 3.1.2 RDD . . . 14 3.1.3 Cluster mode . . . 15 3.1.4 SparkSQL . . . 16 3.2 Amazon S3 . . . 17
3.2.1 Integrazione: connettore Hadoop S3A . . . 17
3.3 Apache Parquet . . . 17 3.4 Apache Zookeeper . . . 19 3.5 Elasticsearch . . . 20 3.6 Dremio . . . 21 3.7 Docker . . . 22 3.7.1 Dockerfile . . . 23 3.7.2 Docker-compose . . . 24 3.7.3 Docker-Hub . . . 24
4 Raccolta e analisi delle metriche di produzione 25 4.1 Specifica dei requisiti . . . 25
4.1.1 1. Il sistema deve acquisire, rielaborare e salvare i dati delle macchine . . . 26
4.1.2 Salvataggio dei dati . . . 26
4.1.3 2. . . 27
4.2 Struttura dei dati acquisiti dall’esterno . . . 27
4.2.1 Esempi di sintassi . . . 28
4.2.2 Obiettivi . . . 29
5 Modello di infrastruttura e elaborazioni 31 5.1 Batch layer: schema dell’infrastruttura . . . 31
5.2 Storage system . . . 32
5.2.1 Data model . . . 33
5.2.2 Componente di storage e formato dei file . . . 35
5.3 Processing and analysis engine . . . 35
5.4 Elaborazione sul batch layer . . . 36
5.4.1 Append sul master dataset . . . 37
5.4.2 Interpolazione delle metriche . . . 39
5.4.3 Esempio di elaborazione di una aggregazione complessa: conta degli stop . . . 41
6 Implementazione dell’infrastruttura 43 6.1 Uso di Docker . . . 43
6.2 Batch Layer: infrastruttura . . . 44
6.3 Amazon S3 come storage Service . . . 45
6.3.1 Permessi . . . 45
6.3.2 Integrazione con Spark . . . 45
6.4 Cluster di Apache Spark . . . 46
6.4.1 implementazione . . . 47
6.5 High-availability . . . 48
6.5.1 Apache Zookeeper . . . 49
6.5.2 Implementazione . . . 49
6.6 Dremio come abstraction layer . . . 50
6.6.1 implementazione . . . 50
6.7 Elaborazioni del batch layer . . . 51
6.7.1 Strumenti utilizzati . . . 51
6.7.2 Operazione di append sul dataset . . . 52
6.7.3 Interpolazione delle metriche . . . 58
6.7.4 Esempio di elaborazione di un’aggregazione complessa: conta degli stop . . . 61
6.7.5 Considerazioni . . . 66
7 Esperimenti e validazione 67 7.1 Test . . . 67
7.1.1 Append delle metriche sul master dataset . . . 68
Sommario
L’obiettivo fondamentale di questa tesi `e stato quello di progettare e imple-mentare il batch layer di un’infrastruttura Big Data che segue il modello dell’architettura lambda . E stata definita una pipeline che si serve di un` sistema di Storage-as-a-Service, ovvero un servizio di storage in cloud con le caratteristiche di un file system remoto. Grazie a questa scelta si ha la garanzia di poter stoccare una quantit`a di dati potenzialmente illimitata senza gli oneri che derivano dalla gestione di una componente di storage. Rimangono invece i vantaggi in termini di scalabilit`a, derivanti da un’organizzazione dello storage come file system.
Capitolo 1
Introduzione
1.1
Elaborazione in batch e Data lake
Un dei modelli di architettura per la gestione di Big Data che si sono affermati negli ultimi anni `e stato il lambda architecture. La peculiarit`a di questo mo-dello `e di dividere il sistema in tre layer ben distinti e separati: lo speed layer, il batch layer, e il serving layer (vedi fig. 1.1). L’obiettivo fondamentale di que-sta tesi `e stato quello di progettare e implementare una parte dell’architettura lambda: il batch layer . Al contrario dello speed layer che elabora il dato per presentarlo in real time all’uscita del sistema, il layer di cui ci siamo occupati `e adibito alle elaborazioni che richiedono tempi maggiori e quindi necessitano un’esecuzione in batch.
Figura 1.1: Architettura lambda
La progettazione di un’infrastruttura riguarda la creazione di un sistema a blocchi in cui diverse componenti software comunicano per effettuare l’acquisi-zione, il caricamento, l’elaborazione e il salvataggio dei dati. Il principio che sta dietro a questo modello `e quello di disaccoppiare le varie parti dell’infrastrut-tura, evitando lo sviluppo di un sistema monolitico, e favorendo la scalabilit`a sulle varie componenti: ogni blocco infatti deve poter essere eseguito su un clu-ster, con un numero di nodi che varia da componente a componente, e che sia modificabile nel tempo. Ad esempio, con questo approccio `e possibile valutare se ci sono colli di bottiglia e porre rimedio aumentando il numero di nodi delle componenti coinvolte. Fin dall’inizio lo scopo della tesi `e stato quello di definire un’architettura che si servisse di un sistema di Storage-as-a-Service, ovvero un servizio di storage in cloud che avesse le caratteristiche di un file system remo-to. La motivazione principale di questa scelta dipende dal fatto che l’approccio tradizionale prevede, per il modulo di storage, l’installazione e la configurazione di un file system distribuito (solitamente HDFS) e di tutto il software necessa-rio per gestire i nodi e accedere ai dati, mentre usando un servizio S-a-a-S al crescere dei dati non ci si deve preoccupare n´e di aumentare la memoria secon-daria n´e di aggiungere nodi al cluster che gestisce lo storage. In questo modo si ha la garanzia di poter stoccare una quantit`a di dati potenzialmente illimitata senza gli oneri che derivano dalla gestione di questo componente; rimangono invece i vantaggi, in termini di scalabilit`a, che derivano dall’avere una struttura organizzata come un file system distribuito. Da un altro punto di vista, poter salvare i dati su uno spazio illimitato risulta fondamentale perch´e `e ormai chiaro alle aziende che i dati hanno un grande valore e che conviene conservarli per un tempo indefinito. Lo storage come servizio permette anche alle piccole e medie imprese, che non possono sostenere elevati costi di gestione, la creazione di un Data Lake, un contenitore dove poter salvare tutti i dati, anche quelli che non vengono elaborati subito dopo l’acquisizione ma che possono essere validi per analisi future.
L’infrastruttura progettata `e stata utilizzata su dati provenienti da macchine impiegate nella produzione della carta. Nell’ambito di un progetto sull’industria 4.0, finanziato da un’azienda che produce macchine per il Converting. Queste macchine sono state collegate a un’architettura prototipale che elabora i dati solamente in real time e li presenta su un portale allo scopo di dare informazioni sulla produzione, sullo stato delle macchine, e sugli allarmi che scattano durante la produzione. Questo sistema `e attualmente in funzione anche se era stato sviluppato per essere una demo dimostrativa.
1.2
Industria 4.0: sfide future
Con il termine Industria 4.0 si intende quel processo che, nei prossimi anni, ci condurr`a a una produzione industriale del tutto automatizzata e intercon-nessa. Le nuove tecnologie digitali saranno il fulcro di questa evoluzione, la quale avr`a quattro direttrici di sviluppo: lo stoccaggio di grosse quantit`a di dati (big data, open data, internet of things, machine-to-machine e cloud
com-puting); l’estrazione di informazioni utili tramite analytics e machine learning; il miglioramento dell’interazione uomo-macchina; l’impiego di tecnologie per il passaggio dal digitale al reale (Manifattura additiva, stampa 3D, robotica, e tut-te quelle tut-tecnologie per immagazzinare e utilizzare l’energia in modo mirato, razionalizzando i costi e ottimizzando le prestazioni). La fabbrica della 4a
ri-voluzione industriale sar`a composta da macchine completamente connesse tra loro, che dialogano le une con le altre ed effettuano manutenzione preventiva. Ad esempio, la manutenzione dei macchinari da parte dei macchinari stessi, gra-zie all’IoT, superer`a in qualit`a e velocit`a quella degli esseri umani entro il 2020 mentre i robot lavoreranno a contatto con l’uomo e dall’uomo apprenderanno in modo naturale. [1]
Una sfida cruciale nei prossimi anni, per l’industria, `e rappresentata dallo sfrut-tamento di una mai prima sperimentata mole di informazioni derivanti dai Big Data. Le aziende oggi raccolgono grandi quantit`a di dati da sensori eterogenei che sono dislocati ovunque: nei macchinari utilizzati in azienda, nei sistemi di identificazione dei prodotti, nella gestione dei magazzini e della logistica, nei sistemi di monitoraggio della produzione. Spesso, tali dati vengono usati al so-lo scopo di valutare se una macchina sta funzionando o meno, eppure ci sono tecniche ormai consolidate che permetterebbero di estrarre conoscenze impor-tanti da questi. Queste tecniche, identificate come Data Mining, potrebbero, ad esempio, prevedere il verificarsi di un guasto del macchinario, o programmare la manutenzione in modo da evitare possibili guasti. Oggigiorno, le grandi im-prese sono pi`u consapevoli dell’importanza dei dati come fornitori di preziosa conoscenza e quindi di valore. Contemporaneamente all’acquisizione di questa consapevolezza da parte delle aziende, sono state sviluppate nuove piattaforme Open Source (fra tutte Hadoop e Spark) che permettono di gestire grandi moli di dati su piccole reti di normali computer desktop o di server virtuali. Questo ha permesso anche alle piccole aziende, aventi budget ridotti, di poter usufruire di queste tecnologie. [I4SS]
Capitolo 2
Infrastruttura Big data
2.1
Architettura Lambda
Un dei modelli di architettura Big Data che si sono affermati negli ultimi anni `e stato il lambda architecture. L’idea principale dietro a questa architettura `e di dividere le componenti software in una serie di layer (vedi fig. 1.1):
• batch layer • serving layer • speed layer
I motivi principali di questa suddivisione sono due: dividere la gestione e il salvataggio dei dati in modo che questi siano immutabili, e generare il risultato delle query in batch preventivamente per evitare che a causa della mole di dati ci sia un’eccessiva latenza. Per quanto riguarda il primo punto, risulta fonda-mentale stabilire un formato in cui i dati, chiamati grezzi, vengono salvati in modo incrementale e non vengono mai pi`u modificati. La generazione delle viste invece viene eseguita a partire dai dati grezzi e ha come output un risultato la cui granularit`a viene scelta in base alle esigenze delle query che vengono eseguite sul sistema. Aggregando il risultato di una o pi`u viste si possono ottenere le risposte alle query di granularit`a superiore (vedi fig. 2.1).
Figura 2.1: Uso delle batch view nelle query
Il modo in cui i layer coesistono `e abbastanza semplice da comprendere. Tutto parte dal fatto che query = funzione(tutti i dati). Idealmente questa funzione potrebbe essere eseguita sull’intera mole di dati, cosa che oltre ad impiegare un tempo eccessivo avrebbe dei costi estremamente alti.
batch layer
L’approccio alternativo a calcolare il risultato di una query online `e quello di calcolarlo in anticipo; il risultato dell’esecuzione preventiva della query viene chiamato batch view. In questo modo avremo che:
• batch view = funzione(tutti i dati) • query = funzione(batch view)
In un’architettura lambda, il compito di generare le viste di batch `e svolto dal batch layer che oltre a questo ha il compito di gestire il dataset principale, i cui dati sono immutabili ma sempre in continua crescita.
serving layer
Come si `e gi`a detto, il batch layer genera delle viste mentre il passo successivo `e quello di caricarle da qualche parte, in modo che in modo che sia possibile ”interrogare” il dato mediante delle query. `E proprio il serving layer il ”luogo” predisposto a questo scopo: Esso `e formato da uno storage distribuito che carica le nuove viste di batch disponibili e d`a la possibilit`a di fare letture ad accesso casuale in modo da avere sempre a disposizione i dati pi`u aggiornati (generati dal batch layer).
speed layer
Facendo le dovute considerazioni sulla generazione delle viste, risulta evidente che in questo approccio `e presente una falla: la creazione di una batch view `e un’operazione che comporta una certa latenza e, durante il tempo che im-piega per essere eseguita, nel sistema saranno arrivati nuovi dati. Affinch´e il sistema possa essere considerato real time a tutti gli effetti, bisogna trovare un modo per sopperire a questa mancanza. Questo problema viene risolto dallo speed layer, che si distingue dal batch layer per il fatto di processare solo una piccola finestra di dati, quelli pi`u recenti. Con riferimento alla notazione gi`a utilizzata precedentemente, a questo punto, possiamo dire che realtime view = funzione(realtime view, nuovi dati). Ricapitolando:
• query = funzione(batch view, realtime view) • batch view = funzione(tutti i dati)
• realtime view = funzione(realtime view, dati nuovi) [BIGD]
2.2
Aggregate-Oriented data model
Con data model aggregate-oriented ci si riferisce a un modello di sistemi di storage non-relazionali(NoSQL) in cui i si tende a mettere pi`u campi su un unico record della stessa entit`a anche con campi annidati e si tende a trattarli come una unit`a; principio che va nella direzione opposta rispetto ai database relazionali. `
E un modello usato nei Domain-Driven Design infatti si parla di unit`a per manipolazione e consistenza dei dati. Aggregare i dati permette di ottenere la consistenza e l’atomicit`a anche nei modelli NoSQL e questo semplifica le cose sui database che operano su un cluster dato che l’aggregato `e l’unit`a ideale per la replicazione e lo sharding (partizionamento dei dati su pi`u cluster). In effetti, l’ascesa dei database aggregate-oriented `e stata dovuta in larga parte all’ascesa dei cluster perch´e l’esecuzione su pi`u nodi ha cambiato le regole sia per quanto riguarda lo storage dei dati, sia per la loro elaborazione. Quando si ha un cluster ci sono un po’ di macchine su cui dividere i dati per l’elaborazione e si ha la necessit`a di ridurre il numero di dati da trasferire sulla rete, preferendo che si processino quelli che risiedono sul nodo dove avviene l’elaborazione. Ad esempio le operazioni di join porterebbero sicuramente a dover recuperare dei dati che non si trovano sul nodo in questione; Questo perch´e i dati sono distribuiti. [NSQL]
2.3
MapReduce
MapReduce `e un modello di programmazione ma anche l’implementazione spe-cifica di un sistema distribuito che ha lo scopo di processare e generare Big Data.
Il programmatore una funzione map, che processa una coppia chiave-valore per generare un insieme di coppie chiave-valore intermedie, e una funzione re-duce che rimette insieme tutti i valori intermedi che hanno la stessa chiave associata. I programmi che vengono scritti con questo stile funzionale, sono automaticamente parallelizzati ed eseguiti su cluster di macchine di migliaia di nodi. Il sistema invece si occupa di tutte quelle operazione a un livello pi`u basso:
• il partizionamento dei dati di ingresso
• lo schedulamento dei processi sulle varie macchine • la gestione dei fault
• la gestione della comunicazione fra macchine
2.3.1
Modello di programmazione
Chi sviluppa usando il modello mapReduce esprime le elaborazioni esattamente attraverso queste due funzioni: map e reduce. Nella fase di map si prende un ingresso e si produce una coppia chiave-valore; nella fase di reduce le coppie vengono raggruppate per chiave e passate alla funzione reduce. La funzione reduce accetta una chiave intermedia e un insieme di valori per la chiave, poi li mette insieme per formare un insieme di valori pi`u piccolo possibile.
2.3.2
Esecuzione
Le invocazioni di map sono distribuite su molte macchine partizionando auto-maticamente i dati di ingresso in M porzioni che possono essere processate in parallelo da macchine differenti. Lei invocazioni di reduce sono distribuite par-tizionando lo spazio delle chiavi intermedie in R pezzi tramite un funzione di partizionamento. [MPRD]
2.4
Complessit`
a nelle infrastrutture big data
Nelle infrastrutture Big Data si possono scegliere diverse soluzioni, riguardo all’hosting delle varie componenti, che portano livelli di complessit`a differen-ti. Usare una soluzione in cloud sicuramente comporta la complessit`a pi`u bassa perch´e i provider di servizi non solo forniscono i container virtuali per l’hosting ma anche un ecosistema globale di strumenti avanzati e prodotti soft-ware. I Vantaggi sono: la gestione dell’infrastruttura quasi completamente a carico del provider; larga scelta di fornitori e di soluzioni software sul merca-to; possibilit`a di minimizzare il time-to-market. Svantaggi: l’utilizzo di servizi di diversi fornitori per le varie componenti software potrebbe portare a sistemi sub-ottimali(anche in relazione coi costi sostenuti) mentre affidarsi a un singolo fornitore potrebbe essere vincolante sulle scelte future. Se si sceglie una so-luzione on-premise mantenere la complessit`a del sistema sotto controllo `e importante al punto che qualunque scelta del sistemista dovrebbe essere fatta
valutando questo aspetto. In sistemi simili diventa necessario automatizzare il pi`u possibile le attivit`a di manutenzione, e installare sistemi di monitoraggio e di alert. Una soluzione ibrida si verifica di solito in due situazioni: quando una compagnia sta trasferendo tutti i suoi dati in cloud gradualmente; quando i dati sensibili vengono mantenuti per scelta on-premise mentre tutti gli altri vengono tenuti in cloud. Questa `e l’infrastruttura pi`u difficile da mantenere sebbene qualche volta necessaria. L’interfaccia tra il cloud e e i dati on premise pu`o diventare una sorgente di ampie incertezze oltre ad aggiungere inefficienza al sistema a causa del ritardo della rete.
La progettazione di un’infrastruttura Big Data riguarda la creazione di un sistema a blocchi che pu`o includere diverse componenti software:
1. un’infrastruttura di streaming di eventi
2. un sistema di storage per il salvataggio di eventi in formato grezzo 3. un database per transazioni veloci e aggregazioni
4. un’infrastruttura di elaborazione e aggregazione dei flussi
5. un datawarehouse per la memorizzazione di eventi in forma strutturata per analisi
6. un motore di elaborazione dati e analisi 7. un motore di query SQL
8. tecnologie relative al Machine Learning
Le componenti software elencate sopra sono come blocchi indipendenti che vengono messe insieme per costruire l’infrastruttura. Nella loro implementazio-ne geimplementazio-neralmente si trovano 4 possibili scenari alternativi: implementazio-nel primo, abbiamo una soluzione completamente in cloud ; nel secondo, una soluzione in cloud ma da fornitori di terze parti ; nel terzo, una soluzione non in cloud che viene fatta funzionare su macchine in cloud ; nell’ultimo, una soluzione completamente on-premise.
Quando si costruisce un’infrastruttura Big Data le questioni pi`u rilevanti riguardano: l’implementazione, la configurazione, la disponibilit`a di metriche di monitoraggio, la capacit`a di scaling, gli aggiornamenti del software in pro-duzione, l’affidabilit`a (fallimenti, persistenza dei dati, replicazione, ripristino), disponibilit`a (come `e distribuito il traffico quando il sistema funziona o va gi`u), possibilit`a di testare il sistema.
Capitolo 3
Tecnologie Utilizzate
In questo capitolo vengono descritte le tecnologie utilizzate per lo sviluppo del-l’infrastruttura con un focus sulle caratteristiche principali e quelle rilevanti per gli utilizzi che se ne sono fatti.
3.1
Apache Spark
Spark `e una piattaforma per l’elaborazione distribuita che estende il modello MapReduce sotto alcuni aspetti:
• maggiore efficienza nell’elaborazione • elaborazione in streaming
• query interattive • funziona in local mode
La cosa che rende Spark nettamente pi`u veloce di Hadoop Map&Reduce `e la capacit`a di eseguire operazioni in-memory, cio`e senza dover scrivere sul disco ad ogni step dell’algoritmo. Questa scelta dei progettisti ha fatto s`ı che Spark, gi`a dal momento della sua creazione, per alcuni tipi di job fosse 10-20 volte pi`u veloce di MapReduce. Un altro punto di forza che riguarda Spark `e il fatto che sia progettato per coprire un’ampia variet`a di elaborazioni diverse (batch, algoritmi iterativi, streaming, query interattive) che prima della sua na-scita dovevano essere eseguite in contesti diversi; dall’uso di Spark ne consegue un risparmio di tempo, perch´e si evita di spostare i dati da una piattaforma all’altra, inoltre si pu`o ridurre il numero di strumenti da utilizzare riducendo i costi di manutenzione. Oltre alle cose gi`a descritte rimane un ultimo aspetto importante Spark funziona anche in locale; questo permette agli sviluppatori di sviluppare prototipi facendoli funzionare su un insieme ridotto dei dati. Spark `e stato progettato per essere altamente accessibile e supporta API in Python, Java, Scala, ed SQL.
3.1.1
Un unico stack
Anche se il core di Spark `e un computational engine esso `e integrato con tanti componenti ad alto-livello(Vedi fig. 3.1). Questo comporta un certo numero di vantaggi:
• le librerie ad alto livello traggono beneficio dalle modifiche a basso livello
• costi minimizzati : si mantiene un solo software integrato invece di 5-10 software indipendenti
• possibilit`a di creare un’unica applicazione in grado di combinare differenti modelli di elaborazione sulla stessa piattaforma
Figura 3.1: Stack di Spark
Spark non necessita di Hadoop, ma per avere il supporto per il sistema di storage implementando le API di Hadoop infatti supporta qualunque suo formato di ingresso.
3.1.2
RDD
Il resilient distributed dataset (RDD ), parte del core di Spark, `e un’astrazione che ci permette di lavorare su una collezione di dati distribuiti. Tutte le ope-razioni da eseguire su Spark sono espresse creando RDD, trasformando RDD esistenti o chiamando operazioni su un RDD per calcolare i risultati. In mo-do trasparente al programmatore, Spark distribuisce i dati tra i nodi paralle-lizzando le operazioni da eseguire. Un RDD offre due tipi di operazioni: le Transformations, e le Actions.
Transformation
Sono operazioni su RDD che danno in uscita un nuovo RDD e che non modifica-no. La davvero interessante di queste `e che sono calcolate in modo lazy, ovvero
solo quando viene invocata un’Action su un RDD generato precedentemente da una transformation.
Action
Le action sono le operazioni che ritornano il risultato dall’esecuzione al dri-ver, o lo scrivono sul sistema di storage; inoltre forzano l’esecuzione delle trasformazioni che avevano generato l’RDD su cui vengono chiamate.
Lazy evaluation
Quando su un RDD viene invocata una transformation Spark salva dei me-tadati internamente per indicare che `e stata richiesta un’operazione. Questo ci fa pensare all’RDD pi`u come a una lista di istruzioni da eseguire sui da-ti che come a un semplice contenitore; lo scopo di avere la lazy evaluada-tion `e quello di ridurre il numero di passi da eseguire sui dati raggruppando pi`u operazioni insieme, e permettendo una valutazione su tutta la pipeline. Il vantaggio di questo approccio rispetto a quello che si ha in altri sistemi come Hadoop MapReduce `e evidente: lo sviluppatore non deve preoccuparsi di come suddividere o raggruppare l’esecuzione delle operazioni; lo fa il sistema valutando la strategia pi`u efficiente. Anche se sembrerebbe secondario, come ul-teriore vantaggio la lazy evaluation permette di scrivere un codice pi`u leggibile e modulare perch´e non vi `e la necessit`a di scriverlo in funzione di come verr`a eseguito.
Persistenza(caching)
Poich`e l’RDD `e un oggetto immutabile quando viene invocato un suo metodo il contenuto di questo non cambia; tutt’al pi`u, `e il risultato del metodo invocato ad essere un nuovo RDD se si sta invocando una trasformazione. In certe situazioni potrebbe succedere di dovere invocare metodi diversi sullo stesso RDD creando due flussi separati: in questo caso anche le operazioni a loro comuni verranno eseguite pi`u volte. Per evitare questo fenomeno il sistema d`a al programma-tore, la possibilit`a di usare un meccanismo di persistenza . Con l’invocazione del metodo persist() sull’RDD si ottiene che il risultato delle esecuzioni da effettuare su di esso verr`a mantenuto in cache; tuttavia il metodo persist() non forzer`a l’esecuzione ma verr`a eseguito lazy come le per le transformation. Col metodo unpersist() sar`a invece possibile, liberare la cache.
3.1.3
Cluster mode
Nella cluster mode Spark usa un’architettura master/slave dove si ha un coordinatore centrale e tanti worker distribuiti (vedi fig. 3.2). Il coordinatore viene chiamato driver mentre i processi che eseguono le operazioni sui wor-ker prendono il nome di executor . Il driver e gli executor terminano insieme all’applicazione. Quando l’applicazione `e lanciata su un cluster di macchine si utilizza un servizio chiamato cluster manager ; Spark ne ha uno integrato:
lo stand alone cluster manager ma si possono usare anche software esterni come Hadoop Yarn e Apache Mesos.
Figura 3.2: Cluster di Spark Driver
Il driver `e il processo in cui viene eseguita la funzione main() dell’applicazione; si occupa di creare lo SparkContext, gli RDD ed esegue le tranformation e le action. Oltre a questo converte il grafo delle operazioni da eseguire (il Direct acyclic graph, DAG) in un insieme di stage, aggregando le transformation e applicando le ottimizzazioni possibili. Gli stage a loro volta vengono convertiti in task, la pi`u piccola unit`a di lavoro presente in Spark, che vengono ripartiti fra i vari nodi del cluster. Una volta che ha inviato i task di un’applicazione agli executor, il driver mantiene sempre una vista completa degli executor per tutta la durata dell’esecuzione.
Executors
Gli executor sono processi dei worker, responsabili per l’esecuzione dei task in un determinato job. Gli esecutori vengono lanciati all’inizio dell’applicazione e solitamente terminano insieme ad essa. In alcuni casi potrebbero fallire; Spark comunque riesce a portare avanti l’esecuzione dell’applicazione riassegnando i task dell’executor che ha fallito ad un altro. [LSPK]
3.1.4
SparkSQL
SparkSQL `e un’interfaccia di Spark per lavorare con dati strutturati e dati semi-strutturati. i dati strutturati sono quelli che hanno uno schema, cio`e un insieme di campi conosciuti per ogni record. Con questo tipo di dati, Spark fornisce tre possibili utilit`a:
2. esecuzione di query usando SQL, sia internamente a Spark sia dall’esterno tramite JDBC/ODBC
3. Dentro a un programma Spark scritto in Python, Java, Scala fornisce una ricca integrazione esponendo delle funzioni simili a quelle dell’SQL. Dataset e dataframe
Quando si eseguono query SQL, o si accede in input a dati strutturati o semi-strutturati tramite l’API SQL di Spark il risultato restituito `e un dataset (o un dataframe). Un dataset `e una collezione distribuita di dati mentre Dataset API `e un’interfaccia che riesce a integrare i vantaggi dell’RDD con i vantaggi di SparkSQL ma `e disponibile solo per Scala e Java. Su Python non `e supportata e neanche su R ma grazie alla natura dinamica di questi linguaggi molti dei benefici di Dataset API sono gi`a disponibili. Un dataframe, invece, `e un dataset organizzato in colonne etichettate concettualmente equivalente alla tabella di un database relazionale o a un dataframe di Python/R ma con grosse ottimizzazioni sotto al cofano. [4]
3.2
Amazon S3
Amazon Simple Storage Service(Amazon S3) `e un servizio di storage a oggetti che offre scalabilit`a, sicurezza, disponibilit`a e durabilit`a dei dati, gestione degli accessi.[5]
3.2.1
Integrazione: connettore Hadoop S3A
Per poter utilizzare lo storage service Amazon S3 nel contesto dei Big Data `e sta-to sviluppasta-to il connetsta-tore Hadoop S3A. Quessta-to ha una serie di caratteristiche che ci permettono di raggiungere lo scopo:
• scrive e legge S3 Objects direttamente
• supporta upload partizionati quando i file sono pi`u grandi del limite del servizio
• Offre l’accesso casuale ai blocchi dei file parquet per migliorare le prestazioni
• supporta la gestione degli accessi e dei ruoli [6]
3.3
Apache Parquet
Apache Parquet `e un formato binario creato appositamente per i big data e utilizzato con Hadoop nei contesti pi`u disparati. `E un formato che ha l’organiz-zazione logica di una tabella, per cui i dati stanno su righe e colonne. Questo
accade solo logicamente, in quanto un file Parquet a livello strutturale ha una gerarchia che si articola su diversi livelli. Da quello pi`u esterno a quello pi`u in-terno abbiamo: RowGroup, Column, Page. Un RowGroup `e un gruppo che contiene una parte dei record (o tutti) e al suo interno `e suddiviso in Column. Per spiegare come sono organizzate le Column dentro al RowGruop facciamo un esempio: mentre gli attributi che su una tabella si trovano sulla stessa linea su un file csv occupano porzioni di memoria contigue, su un file parquet sono le colon-ne all’interno di un RowGroup a essere continue in memoria; quando terminano i record di una colonna inizia la colonna successiva degli stessi record. Anche le Column hanno al loro interno ulteriori strutture, le Page. Strutturalmente parlando, una Page contiene una frazione dei dati che sono contenuti dentro a una Column. Il vero motivo che ne giustifica l’esistenza riguarda il fatto che una Page `e un’unit`a indivisibile in termini di codifica e compressione, inol-tre la suddivisione in Page ci d`a la possibilit`a di richiedere un sottoinsieme dei dati evitando di recuperare quelli che non servono per l’operazione da eseguire. Come si pu`o vedere in fig. 3.3, nel footer di un file parquet possiamo trovare metadati che contengono informazioni relative a tutte le Page. Come metadato fondamentale abbiamo, ad esempio, il tipo di compressione usata per ogni Page ma ci sono altri metadati che contengono statistiche capaci di ridurre i tempi delle SELECT. Supponendo di voler recuperare tutti i valori al di sotto di una soglia, utilizzando il metadato min potremmo scartare le pagina che hanno al loro interno un valore minimo maggiore della soglia richiesta. [3]
Figura 3.3: Apache Parquet
3.4
Apache Zookeeper
Zookeeper `e un servizio distribuito di coordinamento per applicazioni distribuite. I servizi che fornisce sono:
• gestione delle configurazioni • naming
• gestione dei gruppi
• sincronizzazione distribuita
Zookeeper `e la soluzione per la fornitura dei servizi appena elencati perch´e farlo all’interno di un’applicazione distribuita pu`o essere molto complesso; i servizi di coordinamento sono molto difficili da progettare, soprattutto perch´e nascondono insidie come la corsa critica e i deadlock. Zookeeper espone delle primitive molto semplici che le applicazioni usano per implementare servizi pi`u complessi. Usando Zookeeper si evita di assumersi il rischio di implementare un servizio di coordinamento da zero. Caratteristiche chiave di Zookeeper :
• semplicit`a: Zookeeper organizza le informazioni come su un file system. Solo che non `e progettato per lo storage ma tiene tutti i dati in memoria per essere veloce
• replicazione: anche Zookeeper `e replicato su pi`u macchine come i pro-cessi distribuiti che esso coordina
• ordinamento: Zookeeper tiene un ordine per le transazioni eseguite • velocit`a : zookeeper `e veloce sopratutto nei contesti read-dominant
(al-meno un rapporto 10:1 fra letture e scritture) [7]
3.5
Elasticsearch
Elasticsearch `e un motore di ricerca e di analisi distribuito che si occupa della indicizzazione, della ricerca e dell’analisi di dati. Sia per dati strutturati che non, riesce a salvare e indicizzare testo, dati numerici, dati geospaziali in modo che siano supportate ricerche veloci .
Documenti e indexing
Elasticsearch pu`o anche essere visto come uno storage di documenti distribuito che, invece di salvare le informazioni come le righe e le colonne di una tabella, salva strutture dati complesse, serializzate come JSON. Quando `e implementato come un cluster i documenti salvati vengono distribuiti fra i suoi nodi e possono essere acceduti immediatamente da qualunque di loro. Da momento in cui un documento viene salvato, nell’arco di un secondo `e gi`a stato indicizzato e pu`o essere ritrovato tra i risultati di una ricerca. Per implementare le query full-text Elasticsearch usa una struttura a indice invertito in cui ogni parola compare solo volta ma punta a tutti i documenti che la contengono. Mentre un indice pu`o essere pensato come una collezione di documenti inseriti in modo ottimizzato, ogni documento pu`o essere pensato come una collezione di coppie chiave-valore che formano i campi di un oggetto. Di default ogni campo di un documento viene indicizzato perch´e in un indice differente, tutto allo scopo di eseguire le query pi`u velocemente possibile. Elasticsearch `e schema-less, infatti i documenti posso essere indicizzati senza definire esplicitamente il tipo dei dati; sar`a Elasticsearch a mappare i tipi dinamicamente, man mano che vengono inseriti documenti con nuovi campi.
Ricerca e analisi
La vera forza di Elasticsearch sta nelle sue funzioni di ricerca e analisi. Esso fornisce una API REST che supporta query in JSON che seguono Query DSL(Domain Specific Language) inviate dentro al body delle richieste http. Con le query si possono fare ricerche utilizzando un variegato insieme di criteri di
selezione dei dati, e contemporaneamente si possono richiedere delle aggregazioni rispetto a uno dei campi dei documenti.
Figura 3.4: Nodi e shard Cluster, nodi e shard
Elasticsearch `e costruito per far fronte alle esigenze di scalabilit`a e disponibi-lit`a infatti `e possibile aggiungere dei nodi al cluster per aumentare la capacit`a mentre la gestione dei dati `e lasciata al controllo del motore di ricerca. Sotto il cofano, un indice di Elasticsearch, `e solo un raggruppamento logico di shard , un indice autonomo. Distribuendo i documenti in un indice fra pi`u shard, e distribuendo gli shard fra i nodi, si ottiene la ridondanza che serve a fare fronte a due bisogni: il fallimento dell’hardware e l’aumento della capacit`a del-le query (quest’ultimo insieme all’incremento dei nodi). Esistono due tipi di shard : primari e repliche(vedi fig. 3.4); uno shard replica `e la copia di uno shard principale e serve a fornire copie ridondanti dei dati per gli scopi descritti pocanzi. Il numero di shard primary di un indice, deve essere fissato durante la creazione dell’indice e non pu`o essere pi`u cambiato senza che venga reindicizzato tutto, mentre il numero delle repliche pu`o variare. [8]
3.6
Dremio
Dremio `e un prodotto software che i suoi ideatori definiscono Data Lake Engine e che si poggia direttamente su uno storage service senza l’uso di un interme-diario come un process engine per ricavare i dati richiesti. Esso permette di fare delle query direttamente sui dati e ha un meccanismo interno di cache che si serve dei file parquet per accelerare le query che vengono richieste inte-rattivamente. Dremio permette di creare le Data Reflection, strutture dati ottimizzate fisicamente che possono rendere alcune query particolarmente pi`u veloci. Questa cosa viene fatta da Dremio in modo trasparente infatti prima di eseguire una query controlla tutte le Data Reflection generate e usa quelle che possono migliorare le prestazioni della query. A tutti gli effetti si pu`o dire che Dremio fa una copia dei dati solo che di essi mantiene solo il contenuto: per la
struttura usa un modello che gli porta vantaggi in termini di prestazioni. Un altro vantaggio derivato dall’uso di Dremio `e la possibilit`a di connettere qualun-que sorgente di dati (O comunqualun-que supporta buona parte dei servizi di storage esistenti). Un’altra sua peculiarit`a `e quella di riuscire a creare delle tabelle vir-tuali dai risultati delle query appena eseguite. In pratica ogni volta che viene aperta questa tabella sotto il cofano il sistema esegue la query complessa; inoltre `e possibile salvare pi`u volte le tabelle aggiungendo o rimuovendo comandi SQL dalla query. Una cosa che finora non `e stata detta riguarda il linguaggio con cui si possono scrivere le query attraverso una web gui: infatti Dremio permette all’utente di utilizzare SQL abilitando anche gli utenti con meno skill di infor-matica a poter eseguire le query. Infine Dremio riesce a lavorare su un cluster e si serve di Apache Zookeeper per poter dialogare con gli altri nodi.
3.7
Docker
Figura 3.5: stack dei container
Un container `e un’unit`a standard di software che impacchetta il codice di un’ap-plicazione e tutte le sue dipendenze in modo che possa essere eseguita velocemen-te da un ambienvelocemen-te di calcolo a un altro. Un container di Docker, in particolare ha le seguenti caratteristiche:
• standard: Docker ha creato uno standard nell’industria per i container cosa che gli consente una portabilit`a completa
• lightweight: I container condividono il kernel del sistema Operativo della macchina su cui girano; per questo motivo non richiedono un OS per ogni container applicativo portando a una maggiore efficienza del server e riducendo i costi di licenza
• secure: Docker fornisce di default l’isolamento necessario a rendere i container delle entit`a ben divise dal sistema della macchina
Un’immagine Docker `e un pacchetto eseguibile, leggero e autonomo che in-clude tutto ci`o che gli serve per eseguire l’applicazione: codice, runtime, strumenti di sistema, librerie di sistema e impostazioni. Le imma-gini dei container diventano container a tutti gli effetti quando vanno in ese-cuzione sul Docker Engine. I container isolano il software dall’ambiente in cui girano in modo da assicurare che lavorino in modo uniforme a prescindere dall’infrastruttura su cui si poggiano. [9]
3.7.1
Dockerfile
Docker pu`o costruire immagini in modo automatizzato leggendo le istruzioni da un file apposito chiamato Dockerfile. Esso `e un documento di testo che contiene tutte le istruzioni che un utente potrebbe chiamare da riga di comando per assemblare un’immagine. Usando il comando docker build gli utenti potranno eseguire i comandi presenti sul Dockerfile. SOlitamente un Dockerfile nella sua intestazione principale si riferisce al nome di un’immagine esistente sul cloud o in locale per poter estendere le sue caratteristiche:
• aggiungere i propri software e file all’immagine
• installare librerie tramite i comandi della shell di posix • creare utenti e impostare la directory di lavoro • aggiungere variabili d’ambiente
• impostare limiti sull’utilizzo delle risorse da parte dei container
• impostare il processo da mandare in esecuzione sul container (che una volta terminato fa terminare anche l’esecuzione del container stesso) • esporre delle porte sulle reti virtuali a cui un container che esegue questa
immagine appartiene [10]
3.7.2
Docker-compose
Compose `e uno strumento per definire e mandare in esecuzione applicazioni Docker formate da pi`u container. Una volta scritto il file di configurazione chia-mato docker-compose.yml con un singolo comando `e possibile creare o avviare tutti i container necessari. Docker-compose vede il suo maggiore utilizzo quan-do si vogliono avviare pi`u container appartenenti alla stessa applicazione con una configurazione unificata; nulla toglie per`o che si possa utilizzare anche come strumento in cui impostare le configurazioni di un singolo container. Docker-compose unifica tutte le operazioni per la creazione e/o l’avvio di container, di seguito alcune:
• avvio multiplo di container
• impostazione delle dipendenze fra container
• impostazioni relative alla terminazione di un container (es. restart ) • impostazioni dei nomi dei container
• impostazione di Dockerfile alternativi
• mappaggio di porte sulla macchina fisica, o loro esposizione sulle reti virtuali
• mappaggio di volumi e cartelle della macchina fisica dentro ai container • inserimento in una rete virtuale
Usando Compose il deploy di un’applicazione pu`o essere fatto in soli tre passi: 1. creazione di un Dockerfile in modo da definire un sistema riproducibile
in qualunque contesto
2. definizione dei servizi all’interno del file docker-compose 3. esegui docker-compose up per creare e avviare i container
[11]
3.7.3
Docker-Hub
Docker gestisce le immagini che gli utenti generano e voglio rendere pubbliche, in modo unificato, su una piattaforma chiamata Docker-Hub. Queste immagini sono costruite con l’obiettivo di integrare al loro interno solo il minimo software necessario allo scopo desiderato per cui sono presenti milioni di immagini tra cui scegliere e selezionare quella che aderisce maggiormente alle proprie neces-sit`a. Un altro aspetto positivo sono le pagine descrittive che spesso (quando esistono) mettono lo sviluppatore nella condizione di avere un deploy agile. La piattaforma, ad oggi, contiene quasi tre milioni di immagini . [12]
Capitolo 4
Raccolta e analisi delle
metriche di produzione
Ci troviamo nel contesto della produzione della carta tramite semilavorati, il Tissue, nell’ambito di un progetto di Industria 4.0 finanziato da un produttore di macchine per il Converting . Lo scopo del progetto `e quello di informatizzare i processi di produzione, in modo da creare un sistema di generazione e visualiz-zazione di diversi indicatori di produzione. Vi `e la necessit`a di ricavare delle metriche da diversi sensori presenti sulle macchine, raccoglierli, rielaborarli e salvarli. Una volta salvati, da questi dati devono essere ricavati degli indicatori di produzione. Ai clienti deve essere data la possibilit`a visualizzare gli indicatori di produzione ma anche di richiederne diversi in modo interattivo. Il sistema deve riuscire a gestire un numero di macchine che potenzialmente cresce nel tempo e un flusso continuo di dati in ingresso. Il sistema deve salvare i suoi dati in uno storage service in cloud.
4.1
Specifica dei requisiti
Tralasciando i requisiti che devono soddisfare le macchine noi ci occuperemo soltanto dell’analisi dei requisiti del sistema.
Requisiti del sistema:
1. il sistema deve acquisire, rielaborare e salvare i dati delle macchine 2. il sistema deve generare determinati indicatori di produzione
3. il sistema deve permettere all’utente di visualizzare gli indicatori di pro-duzione generati
4. il sistema deve permettere all’utente di richiedere degli indicatori di pro-duzione interattivamente
Requisiti NON funzionali: 25
5. il sistema deve riuscire a gestire un numero di macchine che potenzialmente cresce nel tempo
6. riuscire a gestire un flusso costante di dati in ingresso
7. il sistema deve salvare i suoi dati in uno storage service in cloud
Facendo parte di un progetto pi`u ampio ci sono degli aspetti che in azienda sono gi`a stati affrontati e che di conseguenza non sono stati trattati nell’ambito di questa tesi. Soddisfare il requisiti 2, ad esempio, non rientra nei nostri scopi.
4.1.1
1. Il sistema deve acquisire, rielaborare e salvare i
dati delle macchine
Le operazioni descritte da questo requisito devono essere eseguite in modo atomico, pena la mancata efficacia dell’operazione nella sua interezza.
Acquisizione dei dati
Con questo requisito si intende la capacit`a di acquisire i dati grezzi che le mac-chine producono. A livello infrastrutturale si pu`o tradurre nella capacit`a del si-stema di connettersi a sistemi di storage esterni o di ricevere i dati direttamente dalle macchine o da altri sistemi.
Rielaborazione dei dati
Con questo requisito si intende la capacit`a del sistema di poter rielaborare i dati acquisiti allo scopo di effettuare operazioni di ripulitura prima di salvarli. A livello infrastrutturale si traduce con la capacit`a di calcolo, la capacit`a di scalare con le risorse per fare fronte alla mole dei dati.
4.1.2
Salvataggio dei dati
Con questo requisito si intende la capacit`a del sistema di salvare i dati rispet-tando i principi di replicazione per fare in modo di evitarne la perdita.
4.1.3
2.
Figura 4.1: Infrastruttura attuale
L’infrastruttura attuale non segue il paradigma della lambda architecture n´e tanto meno prevede che ci siano delle elaborazioni in batch. Le operazioni eseguite possono essere riassunte col diagramma di flusso in fig. 4.2. Lo scopo dell’architettura attuale `e quello di eseguire in real time tutte le trasformazioni sui dati necessarie alla dashboard del sistema per mostrare le metriche rilevanti e il loro andamento nel tempo. Sebbene di grande interesse, la descrizione dettagliata di questi task esula dagli obiettivi principali di questa tesi.
Figura 4.2: Data Flow
4.2
Struttura dei dati acquisiti dall’esterno
I dati inviati dalle macchine al sistema riguardano parametri di produzione, informazioni sul loro stato, sulla produzione corrente, e sui diversi allarmi che scattano durante la produzione. Come convenzione, inviano un pacchetto me-diamente ogni 4 secondi, e in linea di massima trasmettono solo i valori delle
metriche che subiscono una variazione. Oltre a queste, viene inviata una me-trica fittizia, keepAlive che permette al sistema di valutare se la macchina `e ancora in linea; a tale scopo viene inviata ogni 4 secondi di default (e non sol-tanto quando subisce una variazione). Ogni record dei dati presenta i seguenti attributi:
• l’attributo Timestamp, stringa di caratteri numerici che mantiene l’in-formazione temporale nel formato YYYY-MM-DD HH:MM:SS.
• l’attributo Metric, stringa di caratteri che con una sintassi specifica d`a informazioni su factory, job, machineType, metric. Con factory ci si rife-risce a uno stabilimento specifico, con job al codice di installazione della macchina in un impianto e con machineType alla tipologia di macchina, seguita da un numero di sequenza nel caso ci fossero due o pi`u macchine (dello stesso tipo) associate allo stesso job. Per finire, metric rappresenta il nome della metrica rilevata. La coppia (job, machineType) identifica univocamente una specifica macchina all’interno del sistema.
• l’attributo Value, stringa di numeri che esprime, come decimale, il valore della metrica indicata nell’attributo Metric.
Per ogni macchina, nella condizione di funzionamento ideale, viene recuperata, in condizione di funzionamento ideale, almeno una metrica per timestamp (ogni 4 secondi). Le macchine, infatti, non comunicano ad ogni timestamp il valore di tutte le metriche ma si limitano a inviare quelle che subiscono una varia-zione. La metrica di clock ha, invece, lo scopo di mostrare che la macchina `e ancora collegata o attiva nelle situazioni in cui i valori delle altre metriche non cambiano. Per questo motivo viene inviata ad ogni intervallo.
4.2.1
Esempi di sintassi
Per dividere logicamente le informazioni principali sono stati usati i punti o i trattini bassi. Eseguendo uno split della metrica rispetto agli opportuni segni di interpunzione vengono fuori delle sottostringhe:
• sintassi principale: la sottostringa centrale serve a identificare una mac-china univocamente mentre la prima e l’ultima danno, rispettivamente, informazioni su stabilimento e nome della metrica.
• sintassi alternativa 0: sono utili solo la prima e l’ultima sottostrin-ga. Dalla prima si estraggono stabilimento, job, e tipo della macchina, dall’ultima il nome della metrica.
• sintassi alternativa 1: ogni sottostringa fornisce una sola informazione. Le sintassi alternative 2, 3 e 4 vengono fatte convergere nella sintassi prin-cipale: nella 2 viene scartato il numero preceduto dalla lettera (nell’esempio T2103); nella 3 viene scartato il numero successivo alla macchina; nella 4 vengono rimosse le lettere che precedono il job (PA).
Tipo di Sintassi Timestamp Metric Value PRINCIPALE 2019-05-08 00:00:00 STABILIMENTO1.44444
REW TISSUEDATA .STATUS CLOCK
30.0
ALTERNATIVA 0 2019-05-08 00:00:00 STABILIMENTO 33333 BUND 8766.BUND 8766.DATA.unit/ .fTDC DataExchange 1 0 1 .gTDCbundler .boStopByEntry 30.0 ALTERNATIVA 1 2019-05-08 00:00:00 STABILIMENTO.22222.WRAP2 .i32ProductLife 30.0 ALTERNATIVA 2 2019-05-08 00:00:00 STABILIMENTO .11111 T2103 REW TISSUEDATA .STATUS CLOCK 30.0 ALTERNATIVA 3 2019-05-08 00:00:00 STABILIMENTO.55555 LS 112 TISSUEDATA.STATUS CLOCK 30.0 ALTERNATIVA 4 2019-05-08 00:00:00 STABILIMENTO.PA11111 REW
TISSUEDATA.STATUS STOPBYFAULT
30.0
Tabella 4.1: sintassi delle metriche
4.2.2
Obiettivi
Va sviluppata un’infrastruttura per il contesto appena descritto che sia in grado di processare i dati acquisiti in modo da fare il parsing delle metriche che le macchine inviano al sistema; le metriche interpretate devono essere salvate in modo definitivo, cio`e una volta e per sempre. Da questo insieme di dati immu-tabili ed eterei devono essere generate delle viste sulle quali poi saranno eseguite le query; trovandoci nel contesto dei big data, le query potrebbero impiegare troppo tempo per essere eseguite on-line quindi per garantire una velocit`a di esecuzione utile allo scopo ha senso generare i risultati prima che vengano ri-chiesti. Il flusso dei dati appena descritto (vedi fig. 4.3) segue un paradigma che aderisce perfettamente al modello definito dall’architettura lambda. La strategia aziendale per lo speed layer `e gi`a consolidata. Quello che invece l’azienda ha necessit`a di approfondire e su cui si incentra la tesi `e il batch layer : uno dei punti pi`u delicati della progettazione di un’infrastruttura per il trattamento di grosse quantit`a di dati. Un altro punto importante per l’azienda `e l’utilizzo di uno storage service in cloud per valutare la fattibilit`a di un’infrastruttura in cui si evita di utilizzare hdfs.
Capitolo 5
Modello di infrastruttura e
elaborazioni
Come gi`a detto precedentemente, l’architettura Big Data che `e stata presa come modello `e la lambda architecture. Lo scopo di questo lavoro di tesi `e stato quello di progettare e implementare il batch layer per un sistema che gestisce i dati provenienti da macchine industriali.
5.1
Batch layer: schema dell’infrastruttura
Una qualunque infrastruttura Big Data `e formata da diverse componenti, ognu-na delle quali ha uno scopo ben definito e separato dalle altre. Per quanto riguarda il batch layer i blocchi necessari nella nostra infrastruttura sono quelli mostrati in figura 5.1: il distributed configuration service, il processing and ana-lysis engine, lo storage system, l’ abstraction and query acceleration engine, il workflow scheduler.
Figura 5.1: Batch layer: infrastruttura
Distributed configuration service
Un servizio distribuito di configurazioni serve a mantenere le configurazioni che i nodi di un cluster gli inviano per essere tolleranti a possibili fault .
Cluster manager
Per andare incontro a un numero crescente di dati il batch layer deve riuscire a scalare le proprie risorse, da uno fino a qualche a migliaio di nodi. Per permettere questo `e necessario l’utilizzo di un componente software che viene eseguito su uno nodo centrale detto master e che coordina tutti gli altri nodi detti slave, i quali, a loro volta, eseguono una processo apposito, per svolgere il loro di nodi periferici.
Processing and analysis engine
Il motore per l’elaborazione e l’analisi `e il componente che esegue tutte le elabo-razioni: i processi di trasformazione dei dati per il loro salvataggio nel master dataset e la creazione delle batch view.
Storage system
Lo storage `e il componente che, per il batch layer, solitamente ha la forma di un file system. Questo gli permette di offrire una vista dei dati come file; inoltre gli permette di partizionare i dati su pi`u file, scegliendo il criterio pi`u conveniente agli scopi da portare a termine. Per eseguire il partizionamento si divide il dataset in cartelle, il nome di ognuna di queste `e composto dal nome del campo pi`u il valore che esso assume in tutti i file contenuti al suo interno. Se per il partizionamento vengono usati pi`u attributi i dati hanno una profondit`a di annidamento rispetto alla radice pari al numero di campi scelti.
Abstraction and query acceleration engine
Questo componente serve a dare un accesso standard ai dati, a prescindere dal formato con cui sono stati salvati. Alcune di queste componenti software riescono anche a unificare l’accesso a dati provenienti da sorgenti diverse, o aventi formati di file differenti; oltre al vantaggio di avere un formato di uscita comune e una vista in tabelle.
5.2
Storage system
Il batch layer di un’architettura lambda segue fondamentalmente un principio inviolabile, l’immutabilit`a dei dati (Vedi 2.1). Il master dataset `e la sede dei dati grezzi, quei dati sui quali oltre all’operazione di lettura pu`o essere ese-guita soltanto un’altra operazione, quella di append. Solo il batch layer deve essere in grado di scrivere nel o leggere dal master dataset e a lui `e demandata
la manipolazione dei dati acquisiti in modo che abbiano il formato scelto per l’inserimento.
5.2.1
Data model
Una parte fondamentale della progettazione di un’infrastruttura `e la scelta del modello dei dati; a maggior ragione in questo caso in cui una volta inseriti sul master dataset non si avr`a la possibilit`a di modificarne la struttura. Un primo schema entit`a-relazione partendo dai dati acquisiti pu`o essere quello in fig. 5.2.
Figura 5.2: Schema relazionale dati acquisiti
Questo schema E-R mette sullo stesso piano tutte le metriche senza consi-derate che alcune potrebbe avere tipi completamente diversi fra di loro. Questo pu`o creare problemi qualora si volesse usare una struttura dati che ha uno sche-ma rigido (in questo caso per schesche-ma si intende la definizione dei tipi degli attributi dei record). Per modificare lo schema attuale ci si trova a un bizio: il giusto compromesso tra spazio occupato e complessit`a dello schema. Due possibili soluzioni considerate sono:
• un’entit`a per ogni metrica (fig. 5.3)
Figura 5.3: Schema E-R con le metriche come entit`a
Figura 5.4: Schema E-R con le metriche come attributi
Per quanto riguarda le metriche la soluzione scelta `e stata la seconda (fig. 5.4). Un aspetto negativo di questo modello `e che il pi`u delle volte buona parte delle metriche sar`a un campo vuoto, infatti, come si era gi`a detto, solo i valori delle metriche che variano durante l’intervallo di campionamento sono inviati al sistema. Un aspetto positivo invece `e la possibilit`a di recuperare su un unico record lo stato della macchina istante per istante (a meno dei campi vuoti che per`o idealmente hanno come valore, l’ultimo inviato dalla macchina in ordine temporale). Questo tipo di modello in cui si cerca di aggregare pi`u campi possibili viene usato negli storage NoSQL e viene definito Aggregate-Oriented (Vedi Sez. 2.2). Sempre seguendo questo modello, allo schema E-R in fig. 5.4 si pu`o apportare un’ultima modifica: L’inserimento dei dati dell’entit`a Machine, all’interno dell’entit`a MachineState (5.5).
Figura 5.5: Schema E-R i dati macchina per ogni stato della macchina A questo punto, su un unico record avremo tutti i dati che si possono ricevere dalle macchine come metrica pi`u le informazioni che la identificano univocamente e il nome dello stabilimento.
5.2.2
Componente di storage e formato dei file
La tecnologia che viene sempre usata per lo storage dei dati grezzi su un’infra-struttura big data `e il filesystem distribuito. Essa si sposa perfettamente con l’esecuzione in batch in quanto legge e scrive sequenze di byte in blocchi e lo fa nel modo pi`u veloce rispetto a qualunque altro sistema di storage. Mentre l’uso del filesystem `e una cosa assodata, rimane la scelta del formato dei file. In questo caso per i dati grezzi del master dataset si `e deciso di usare Apache Par-quet (Vedi sez. 3.3), un formato di successo sui sistemi distribuiti che include caratteristiche molto importanti come la compressione e il partizionamento. Se adesso ripensiamo al data model e al modello aggregate-based possiamo chiarire un paio di aspetti che potevano non convincere del tutto:
• molti valori null per i campi dei record
• ridondanza degli attributi factory, job, machineType
Questi svantaggi del modello vengono risolti dalle propriet`a del formato. Infat-ti, grazie alla compressione nel caso degli attributi factory, job, machineType, Parquet user`a una sintassi per dire che i record dall’indice 0 all’indice N hanno un valore e non dovr`a ripeterlo per ogni record; stessa cosa per i campi vuoti (se sono consecutivi).
5.3
Processing and analysis engine
Questo componente `e sicuramente il core di un’infrastruttura big data. Le caratteristiche che un processing engine deve avere sono:
• fault-tolerance • generality • high-availability
Il sistema deve riuscire a gestire i dati anche quando la loro mole aumenta e deve farlo incrementando il numero di nodi che l’infrastruttura mette a disposi-zione. La scalabilit`a di un’esecuzione dipende anche dagli algoritmi che vengono sviluppati: se un algoritmo `e parallelizzabile il sistema `e scalabile. Cosa pi`u im-portante dell’essere scalabile per un sistema `e l’essere linearmente scalabile: esso pu`o dirsi tale se, quando aumenta il carico di lavoro, riesce a mantenere le stesse performance con un aumento proporzionale delle risorse. Un sistema che non che non ha quest’ultima propriet`a non `e molto utile perch´e a causa dei costi non sarebbe realizzabile. Poich´e la scalabilit`a, nelle infrastrutture big data, si ottiene attraverso l’utilizzo di un sistema distribuito il nostro processing engine deve essere in grado di eseguire algoritmi in parallelo su pi`u nodi ; ovvero deve essere distribuito sui nodi di un cluster . Un’altra propriet`a importante per un process engine `e la fault-tolerance. Un programma potrebbe fallire per vari motivi: una memoria secondaria potrebbe raggiungere la sua massima capacit`a, il processo potrebbe occupare tutta la memoria a lui dedicata, potrebbe romper-si una componente dell’hardware, etc. . . Per questo motivo, un process engine deve essere in grado di ripartire dopo un fault, rieseguendo soltanto l’elabora-zione sulla porl’elabora-zione dei dati coinvolta. La generality indica che il sistema deve permettere l’esecuzione di qualunque operazione su un’arbitraria porzione dei dati[PRCS]; infine l’ultima propriet`a richiesta, l’high-availability, `e la capacit`a di un sistema di essere sempre pronto a soddisfare le richieste. Per esempio, nei cluster in cui c’`e un solo master non-eleggibile abbiamo un one-point-of-failure per cui se esso va gi`u potrebbe essere necessario far ripartire il sistema manual-mente e il servizio potrebbe non essere disponibile durante il tempo necessario al restart. Dalle propriet`a richieste per questo componente architetturale si arriva alla conclusione che il giusto candidato va cercato fra i sistemi distribuiti che hanno una certa tolleranza ai fault, sono abilitati all’elezione del master, e che sfruttano (anche dietro le quinte) il paradigma Map&reduce con cui distribuire i dati ai nodi, elaborarli, e raccogliere i risultati prima di presentarli in uscita.
5.4
Elaborazione sul batch layer
Le elaborazioni che sono competenza del batch layer sono di due tipi: il primo riguarda l’aggiunta di nuovi dati nel master dataset, il secondo la generazione delle batch view. Il primo tipo di di operazione recupera i dati da un sistema di storage esterno al batch layer, lo manipola e lo aggiunge in append ; il secondo tipo di operazione viene eseguito ciclicamente sull’intero master dataset secondo i principi della lambda architecture (vedi sez. 2.1) Le elaborazioni verranno descritte attraverso dei diagrammi chiamati pipe diagram . Questi diagrammi hanno il pregio di esprimere i flussi di esecuzione etichettando le operazioni con
dei tag che si rifanno al linguaggio SQL, inoltre sono convertibili nel paradigma map&reduce in modo automatico. Questo `e dimostrato anche dal fatto che esistono alcuni framework per l’elaborazione distribuita che stanno a un livello pi`u alto rispetto a quello di map&reduce e permettono di scrivere il codice delle procedure riferendosi alle strutture come se fossero tabelle ed eseguendo funzioni scritte con un paradigma vicino al pi`u noto, se non pi`u intuitivo, linguaggio SQL. I tag che si possono associare ai blocchi del pipe diagram sono i seguenti:
• Function e Filter : le operazioni che presentano questi tag vengono ese-guite su un singolo record alla volta. Per quanto riguarda l’elaborazione distribuita, possono essere eseguite sia in uno step di tipo map che in uno di tipo reduce
• GroupBy : questo tag definisce l’operazione di raggruppamento dei record per gli attributi indicati. Nel paradigma map&reduce corrisponde all’emit di chiavi durante la fase di map
• Aggregators: le operazioni che presentano questo tag indicano come ag-gregare i record che sono stati raggruppati precedentemente. In map&reduce avvengono durante gli step di reduce
• Join: questo tag definisce l’operazione di join rispetto a uno attributo o pi`u
• Merge: le operazioni con questo tag vengono eseguite su pi`u insiemi di dati
[PIPE]
5.4.1
Append sul master dataset
L’operazione che aggiunge nuovi dati al master dataset `e incrementale, ovvero riguarda solo una porzione dei dati acquisiti dal sistema, che non sono stati ancora aggiunti al master dataset. Come stabilito nel Data Model i dati delle macchine in produzione verranno salvati in modo aggregato. L’operazione viene eseguita seguendo la pipe modellata in fig. 5.6.
Figura 5.6: Pipe diagram: aggregazione dati delle macchine
I dati in ingresso all’algoritmo sono quelli inviati dalle macchine, in formato json con la struttura descritta nella sez. 4.2. Come mostrato in fig. 5.6, per ogni json viene fatto il parsing del campo field e viene estrapolato il valore dei campi factory, job, machineType, pi`u il nome della metrica il cui valore `e all’interno del campo value. Insieme all’oggetto, tramite una funzione hash applicata ai campi job, machineType e timestamp, viene generata una chiave con cui viene create una coppia chiave-valore con l’oggetto come valore. Questo diventa l’uscita della funzione di parsing. Nello step successivo le coppie chiave-valore vengono raggruppate per chiave (ordinamento) e vengono aggregate con la funzione di merge. Da tutti gli oggetti il cui parsing `e andato a buon fine, raggruppati per chiave, viene creato un solo oggetto che, condivide con gli oggetti di partenza i valori dei campi job, machineType, timestamp ma che ha anche dei nuovi campi, uno per ogni oggetto di cui si `e fatto il merge. Gli oggetti che non sono stati interpretati correttamente dalle metriche, vengono lasciati immutati e verranno filtrati per essere inseriti in una destinazione diversa. Una volta che i record sono nel formato stabilito non rimane che ordinarli rispetto al timestamp prima
di salvarli.
5.4.2
Interpolazione delle metriche
Le operazioni che vengono eseguite usando i raw data salvati sul master dataset sono quelle che generano le batch view. Una delle batch view che `e utile generare per le serie temporali acquisite in un contento industriale di produzione `e quella che presenta i record arricchiti, laddove un campo era null nel master dataset, con il valore precedente in ordine di timestamp, attuando di fatto un’interpo-lazione delle metriche. Il sistema attualmente in produzione ha un’architettura che funzionando in near-realtime genera costantemente i valori che servono alla dashboard per visualizzare le prestazioni delle macchine. In un’infrastruttura che segue l’architettura lambda i dati devono essere elaborati in blocco tramite il batch layer, mentre lo speed layer potr`a elaborare quelli che non erano presenti nel master dataset all’inizio dell’esecuzione in batch.
Come mostrato in fig. 5.7, i dati in ingresso all’algoritmo sono gli stati delle macchine salvati come raw data nel master dataset mentre i dati in uscita sono gli stati delle macchine con i record interpolati. Il risultato di questo algoritmo va a formare una batch view. Come gi`a detto, l’algoritmo prende in ingresso lo stato delle macchine, poi li raggruppa per job, machineType e aggiunge un attributo il cui valore, per ogni record, sar`a il valore di timestamp subito precedente, in ordine di temporale, a quello preso in considerazione. Nel passo successivo, ognuno dei campi degli attributi da interpolare, viene sovrascritto inserendo, per ogni record, l’ultimo valore valido (NON-NULL) in ordine temporale; questo `e vero nei casi in cui la differenza fra il timestamp del record e quello del record precedente `e minore di una soglia stabilita (che pu`o variare da campo a campo); in caso contrario il valore rimarr`a invariato. Infine, i dati in uscita vengono raggruppati per job, machineType, year, month dopo essersi in modo da essere salvati seguendo le giuste partizioni.
5.4.3
Esempio di elaborazione di una aggregazione
com-plessa: conta degli stop
In ambito industriale le macchine in media lavorano ad una certa velocit`a e a questa corrisponde una quantit`a media di prodotti in uscita dalla linea. In questo contesto ogni stop della produzione ha un costo misurabile quindi si dovrebbe ridurre al minimo il loro numero. Un dato utile da generare come batch view, potrebbe essere il numero di stop giornalieri; in fig. 5.8 abbiamo un pipe diagram che descrive le operazioni effettuate per ottenere questo dato. Le macchine inviano una metrica booleana che segna i momenti in cui il loro stato viene commutato (spenta/accessa o accesa/spenta). Per l’operazione di conta degli stop della macchina, dopo aver raggruppato i dati di ingresso per job, machineType, per ogni record viene recuperato precedente valore di running diverso da NULL in ordine di timestamp; dopodich´e viene controllato se `e uguale a 1 (macchina accesa ) o se `e uguale a 0 (macchina spenta ); infine viene confrontato se il valore di running del record stesso vale 0 (macchina spenta). Se entrambi i confronti danno un risultato positivo allora avremo che una macchina in esecuzione, a un certo punto si `e spenta, ovvero si `e verificato uno stop. Questo risultato viene conservato inserendo un 1 in corrispondenza dei record che riportano lo spegnimento di una macchina. A questo punto, dal timestamp, per ogni record, vengono estratti i valori di anno, mese, giorno, poi per ogni macchina vengono sommati i valori di stop raggruppati per giorno.
Capitolo 6
Implementazione
dell’infrastruttura
6.1
Uso di Docker
Allo scopo di mettere in piedi un’infrastruttura big data formata da cluster vir-tuali che potessero in qualche modo simulare un ambiente distribuito su una sola macchina fisica `e stato scelto di utilizzare Docker. Questo software ci ha permesso, grazie ai container che riescono a virtualizzare in modo efficiente l’ambiente di un sistema operativo, di creare dei cluster tramite i file di confi-gurazione di docker-compose e l’utilizzo di immagini di container presenti sulla piattaforma Docker-Hub (vedi sez. 3.7). I nodi di un cluster comunicano in rete e per fare in modo che questo avvenga per i nodi virtuali di un cluster su Docker, questi devono essere collegati a una rete virtuale tramite un indirizzo IP e possono esporre le porte che vengono richieste. Oltre a questo, alcune porte possono essere mappate su quelle della macchina fisica in base alle necessit`a che si hanno. Una cosa molto comoda per le configurazioni e la loro leggibilit`a `e la possibilit`a di usare i nomi dei container al posto degli indirizzi di rete: sar`a il sistema a preoccuparsi di risolvere gli indirizzi. Anche se in questo caso l’uso di Docker `e stato dettato dalla necessit`a di virtualizzare l’ambiente di una sola macchina fisica, i vantaggi che si traggono dall’uso di questo sistema in generale sono notevoli. Docker permette:
• di replicare velocemente un container su qualunque macchina senza alcuna modifica
• di evitare situazioni impreviste: l’esecuzione di un container su macchine fisiche diverse dar`a lo stesso comportamento
• creare un ambiente isolato per ogni servizio, senza dover gestire dipendenze o conflitti tra servizi diversi sulla stessa macchina fisica
• di impostare dei limiti sull’utilizzo delle risorse 43
• di fare un reset di tutto l’ambiente senza rischiare di tralasciare vecchi file o impostazioni che influenzano il sistema
Queste motivazioni ne giustificano l’utilizzo in qualunque contesto. Anche quan-do si ha la necessit`a di implementare un cluster con diversi nodi fisici per la semplicit`a con cui `e possibile replicare un container.
6.2
Batch Layer: infrastruttura
Figura 6.1: Batch layer: infrastruttura globale
Il Batch Layer di un’infrastruttura che segue il modello della Lambda Archi-tecture (vedi sez. 2.1) `e l’insieme di quelle componenti che fondamentalmente sono adibite a due tipi di operazioni:
• l’acquisizione dei dati da una sorgente esterna e la loro trasformazione per l’inserimento nel master dataset
• la creazione delle batch view dall’intero insieme di dati contenuti nel master dataset
A livello infrastrutturale il sistema costruito `e quello in fig. 6.1 e contiene i seguenti componenti: Apache Spark che integra al suo interno le componenti di processing and analisys engine (core del prodotto software), cluster mana-ger e abstraction query engine; Apache Zookeeper come distributed coordina-tion service; Amazon S3 come storage service; Dremio come abstraccoordina-tion and acceleration query engine.
