{"id":5260,"date":"2022-12-23T09:30:00","date_gmt":"2022-12-23T08:30:00","guid":{"rendered":"https:\/\/blog.besharp.it\/?p=5260"},"modified":"2022-12-23T10:18:00","modified_gmt":"2022-12-23T09:18:00","slug":"etl-serverless-su-aws-integrazione-diretta-sns-kinesis","status":"publish","type":"post","link":"https:\/\/blog.besharp.it\/it\/etl-serverless-su-aws-integrazione-diretta-sns-kinesis\/","title":{"rendered":"ETL Serverless su AWS: Integrazione diretta SNS – Kinesis"},"content":{"rendered":"\n
Al giorno d\u2019oggi, pi\u00f9 che mai, le aziende capiscono il potenziale e il valore dei dati e dell\u2019importanza di utilizzarli in maniera consona. Per questo motivo soluzioni ETL sono sempre pi\u00f9 comuni e dedicate.<\/p>\n\n\n\n
In questo articolo parleremo di una pipeline ETL con un\u2019integrazione non comunemente usata in problemi di questo tipo, ma necessaria per il business preso in esame.<\/p>\n\n\n\n
L’ETL (Extraction, Transformation, and Loading) \u00e8 ormai uno standard per pipeline dedicate al dato.\u00a0<\/p>\n\n\n\n
Lo step di estrazione raccoglie dalle varie sorgenti il dato grezzo: avere un Data Lake correttamente organizzato \u00e8 la chiave per semplificare la fase di estrazione e le successive.<\/p>\n\n\n\n
Una volta che il dato \u00e8 stato recuperato si possono applicare tutte le trasformazioni necessarie ad estrarre il valore maggiore possibile dal dato. <\/p>\n\n\n\n
Questi step in linea, che si riassumono in ci\u00f2 che viene chiamato \u201cstep di trasformazione\u201d, sono specifici in base alle necessit\u00e0 di utilizzo del dato. I risultati saranno passati all\u2019ultimo step di salvataggio: lo step di caricamento (Loading). <\/p>\n\n\n\n
I risultati possono essere analizzati e visualizzati per avere una visione chiara e migliore da un punto di vista analitico in modo da aiutare i processi decisionali.<\/p>\n\n\n\n
Il vantaggio di questa struttura di processo \u00e8 la facilit\u00e0 di modifica e\/o implementazione di una trasformazione aggiuntiva. Il tempo richiesto \u00e8 drasticamente ridotto garantendo cos\u00ec un vantaggio enorme in termini di tempo nel rispondere a domande chiave dal punto di vista business in modo da guidare, dati alla mano, decisioni a pi\u00f9 alto livello.<\/p>\n\n\n\n
Ogni step pu\u00f2 essere realizzato con diverse modalit\u00e0. In questo articolo andremo a vedere una di queste possibilit\u00e0, aggiungendo qualche consiglio riguardo come farlo nel miglior modo.<\/p>\n\n\n\n
Prima di iniziare con la descrizione della soluzione tecnica contestualizziamo il flusso dati.<\/p>\n\n\n\n
L\u2019idea finale \u00e8 un servizio a cui le persone si possano sottoscrivere e inviare flussi di dati continui che dovranno essere salvati e processati in, quasi, tempo reale.<\/p>\n\n\n\n
Il caricamento in maniera corretta, senza perdita di dati, \u00e8 critico per il funzionamento del flusso. <\/p>\n\n\n\n
Inoltre, i componenti dovranno essere in grado di scalare in maniera efficiente.<\/p>\n\n\n\n
Soluzione tecnica: integrazione SNS – Kinesis Firehose e SQS<\/h2>\n\n\n\n
Ora che abbiamo un quadro generale, possiamo iniziare con la descrizione delle componenti della soluzione tecnica, spiegandone le possibilit\u00e0 e come connetterle tra di esse per farle interagire.<\/p>\n\n\n\n
Il servizio AWS SNS pu\u00f2 essere configurato per inviare notifiche con diversi protocolli: dalle classiche eMail e messaggi, alle richieste API HTTP\/HTTPS, fino a integrazioni dirette con servizi AWS come SQS, Lambda e Kinesis Data Firehose.<\/p>\n\n\n\n
Per questo motivo i topic SNS sono un\u2019ottima soluzione per il disaccoppiamento dell\u2019infrastruttura.<\/p>\n\n\n\n
Il topic SNS invier\u00e0, nella nostra soluzione, dati a molteplici destinatari: un Kinesis Delivery Stream e una coda SQS. <\/p>\n\n\n\n
Nota: possiamo anche usare una sottoscrizione email per ricevere direttamente notifiche su un subset di input grazie alle possibilit\u00e0 fornite da SNS filter.<\/p>\n\n\n\n
L\u2019idea \u00e8 processare i dati in input e, al contempo, salvarli in una sezione differente dell’infrastruttura in modo da poterli visualizzare e, se necessario, ri-processarli successivamente con altri tipi di trasformazioni o con una versione aggiornata del nostro flusso dati. <\/p>\n\n\n\n
L\u2019SNS topic si interfaccer\u00e0 quindi sia con la componente di estrazione, che con con quello di trasformazione del nostro flusso ETL: il Delivery Stream salver\u00e0 i dati che faranno parte del nostro Data Lake e la coda SQS invier\u00e0 lo stesso dato al servizio per la trasformazione.<\/p>\n\n\n\n
<\/p>\n\n\n
\n<\/figure><\/div>\n\n\n
<\/p>\n\n\n\n
Kinesis Delivery Stream \u00e8 in grado di aggregare il dato in input e salvarlo come file singolo. Questo \u00e8 esattamente ci\u00f2 che ci serve per smarcare il primo obiettivo.<\/p>\n\n\n\n
Un Delivery Stream aggregher\u00e0 i dati per periodi di tempo configurabili, o fino al raggiungimento di una quota di MBs e salver\u00e0 il file generato in maniera compressa su S3 (.parquet o .orc<\/em>). <\/em><\/p>\n\n\n\n
L\u2019SNS topic indirizzer\u00e0 i dati sia a Kinesis, che al processor; quindi possiamo settare un limite di 3 minuti per il batch dei dati in input. Salveremo file .parquet su un bucket S3 in una cartella \u201cdata\u201d<\/em>, partizionando per anno, mese e giorno con l\u2019espressione:<\/p>\n\n\n\n
La scelta del formato parquet \u00e8 guidata dalla facile accessibilit\u00e0 anche tramite S3 Select, Glue e Athena. <\/p>\n\n\n\n
I file parquet sono strutturati, quindi dovremo avere la definizione di questa struttura da configurare in Kinesis Data Firehose.<\/p>\n\n\n\n
A riguardo, una nota importante: \u00e8 anche possibile avere una funzione Lambda come transformer per il Delivery Stream.<\/p>\n\n\n\n
Lambda pu\u00f2 fare qualsiasi tipo di elaborazione sul dato, dalla semplice validazione alla, completa ricostruzione del dato in input con l\u2019aggiunta di nuovi campi. La definizione della struttura necessaria a Kinesis Data Firehose per l\u2019aggregazione del dato in file parquet deve, per forza di cose, riflettere quella del dato di output che sar\u00e0 salvato in S3.\u00a0<\/p>\n\n\n\n
In questo caso specifico, stiamo usando Kinesis Delivery Stream per salvare il nostro input. Non necessitiamo di una Lambda Transformer e quindi la nostra struttura sar\u00e0 uguale a quella del dato in input.<\/p>\n\n\n\n
La comunicazione della struttura dati a Kinesis \u00e8 fatta tramite i servizi di AWS Glue usando, nello specifico, le feature di Glue Data Catalog. Dobbiamo creare un database Glue con tabelle al suo interno, le tabelle manterranno la struttura dati e saranno usate da Kinesis per la generazione dei file parquet.<\/p>\n\n\n\n
Ora che abbiamo il dato di input salvato e aggregato dobbiamo trovare il modo di processarlo.<\/p>\n\n\n\n
Processare il dato pu\u00f2 essere fatto in diversi modi, come detto in precedenza, i set dei possibili sottoscrittori di SNS \u00e8 vasto. Da funzioni Lambda, code SQS a integrazioni API per notificare tipi eterogenei di processors.<\/p>\n\n\n\n
Nel nostro caso abbiamo necessit\u00e0 di un processing near-real time; abbiamo quindi creato una coda SQS che viene consumata da un servizio ECS. <\/p>\n\n\n\n
Il servizio scala il numero di task accesi dipendentemente dal numero di messaggi all\u2019interno della coda.<\/p>\n\n\n\n
Abbiamo iniziato il flusso con la validazione dell\u2019input controllando che tutti i campi necessari siano presenti. Questo controllo \u00e8 necessario per controllare che nessuno abbia manomesso l\u2019applicazione provando a interagirci in modi malevoli. <\/p>\n\n\n\n
Inoltre, abbiamo inserito dei controlli sul tipo del dato in modo da garantire che l\u2019input sia in linea con i metodi poi utilizzati per il processing. <\/p>\n\n\n\n
Dati validi continueranno all\u2019interno del flusso mentre dati \u2018sporchi\u2019 o non completi saranno categorizzati e, in base al tipo di errore sul dato, salvati in sezioni diverse della soluzione. <\/p>\n\n\n\n
Il processing del dato di input pu\u00f2 essere diviso in 3 parti principali: appiattimento del dato e pulizia, offuscazione e aggregazione. <\/p>\n\n\n\n
Inizialmente appiattiamo il dato portando a livello eventuali dati annidati all\u2019interno di altri filtrando inoltre campi non utilizzati dalla piattaforma. <\/p>\n\n\n\n
Di seguito andiamo a offuscare il dato di input. Non andremo nel dettaglio di questo step ma \u00e8 risultato necessario non avere tutti i dati in chiaro per questo specifico flusso di dati. Abbiamo utilizzato tecniche di data masking per nascondere le PII.<\/p>\n\n\n\n
Come ultimo step aggreghiamo il dato per poi estrarre metriche e generare informazioni da queste ultime.<\/p>\n\n\n\n
Per questa soluzione, siamo interessati nel trovare luoghi fisici in cui persone si ritrovano in gruppi; dobbiamo quindi contare il numero di persone all\u2019interno di un dato raggio con informazioni sul sesso e sull\u2019et\u00e0 dei soggetti.<\/p>\n\n\n\n
Infine, lo step di caricamento del flusso ETL: salviamo il dato aggregato all\u2019interno di un bucket S3 per l\u2019utilizzo futuro.<\/p>\n\n\n\n
Visualizzazione del dato<\/h2>\n\n\n\n
Abbiamo ora il nostro dato, sia processato che grezzo, salvato all\u2019interno di un Bucket S3. Possiamo finalmente utilizzarlo per analisi e raccogliere informazioni.<\/p>\n\n\n\n
Vorremo fare pi\u00f9 cose col dato tra cui: analizzare con query e visualizzarlo.<\/p>\n\n\n\n
Se vogliamo fare query SQL sul nostro dato in S3 possiamo utilizzare Athena.<\/p>\n\n\n\n
Athena pu\u00f2 eseguire query direttamente su S3 o tramite il Glue Data Catalog. Come descritto in precedenza, abbiamo gi\u00e0 creato un database Glue e una tabella all\u2019interno del Data Catalog per i nostri dati in input da Firehose.<\/p>\n\n\n\n
Possiamo fare lo stesso con i nostri dati aggregati utilizzando un altro servizio dalla suite Glue: il Glue Crawler.<\/p>\n\n\n\n
Con poche configurazioni possiamo puntare il nostro Crawler verso una posizione in un bucket S3 e il crawler estrarr\u00e0 in autonomia lo schema del dato creando una tabella Glue che lo rifletta.<\/p>\n\n\n\n
Suggeriamo di seguire questa metodologia in modo che, se cambiamo il tipo di data processing, il crawler potr\u00e0 essere rilanciato e modificher\u00e0 in autonomia lo schema.<\/p>\n\n\n\n
Ora possiamo eseguire query sui nostri dati, sia input che aggregati, con Athena recuperando importanti informazioni di business come, ad esempio, il numero di persone in prossimit\u00e0 di un punto d\u2019interesse, raggruppate in base al sesso.<\/p>\n\n\n\n
<\/p>\n\n\n
\n<\/figure><\/div>\n\n\n
<\/p>\n\n\n\n
La visualizzazione del dato sar\u00e0 relativamente semplice da configurare. In questo caso, ci viene in aiuto AWS QuickSight per la creazione di una dashboard.<\/p>\n\n\n\n
Per farlo, andiamo a creare due dataset in QuickSight, uno per il dato grezzo di input e uno per il processato. I dataset possono avere come sorgente diverse entit\u00e0, nel nostro caso useremo Athena per entrambi.<\/p>\n\n\n\n
Una volta che abbiamo i nostri dataset possiamo utilizzare le analisi QuickSight per personalizzare la visualizzazione come preferiamo e, successivamente, quando siamo soddisfatti del risultato ottenuto, salvarla in una dashboard QuickSight da fornire ai fruitori.<\/p>\n\n\n\n
Grazie alla dashboard QuickSight abbiamo ora un modo di fruire dei risultati delle nostre analisi in maniera chiara ed immediata. Riusciamo quindi ad avere l\u2019informazione dell\u2019aggregazione di persone presso punti di interesse a colpo d\u2019occhio.<\/p>\n\n\n\n
<\/p>\n\n\n
\n<\/figure><\/div>\n\n\n
<\/p>\n\n\n\n
Conclusioni<\/h2>\n\n\n\n
In questo articolo abbiamo esplorato una parte del mondo ETL, con un assaggio di ogni sua componente: estrazione, trasformazione, caricamento ed eventuali implementazioni su AWS.<\/p>\n\n\n\n
Abbiamo imparato come disaccoppiare blocchi infrastrutturali con SNS e come possiamo utilizzare i topic per notificare diversi componenti con lo stesso input.<\/p>\n\n\n\n
Grazie a questa possibilit\u00e0 abbiamo notificato sia lo step di Estrazione, che di Trasformazione nello stesso momento processando cos\u00ec il dato in quasi tempo reale.<\/p>\n\n\n\n
Abbiamo poi descritto le varie possibilit\u00e0 di salvataggio del dato di input in formati compressi, parquet con Kinesis Delivery Stream. Facendo questo, abbiamo esplorato i servizi della suite Glue dal Catalog, con database e tabelle, al Crawler che ci aiutano nell\u2019integrazione di sorgenti esterne con il nostro Data Lake.<\/p>\n\n\n\n
Inoltre abbiamo mostrato come \u00e8 possibile gestire i dati di input con code SQS e task ECS. Abbiamo dato qualche spunto di riflessione sulle tecniche di processing ma, di solito, sono molto specifiche in base al tipo di richiesta business.<\/p>\n\n\n\n
Infine, abbiamo delineato come eseguire query con AWS Athena e come creare e condividere dashboard con QuickSight per la visualizzazione del dato.<\/p>\n\n\n\n
Un\u2019ultima cosa: i flussi di analisi dati sono spesso specifici in base al contesto, sentitevi liberi di sperimentare in ogni step! Da sorgenti di dati differenti, come database, a ogni tipo di trasformazione del dato che potrebbe venirvi in mente; potreste provare con diversi servizi di processing come Lambda che, potenzialmente, potrebbe essere la soluzione pi\u00f9 adatta per alcuni flussi.<\/p>\n\n\n\n
La visualizzazione del dato inoltre \u00e8 lo step in cui si pu\u00f2 sperimentare ancora di pi\u00f9 utilizzando al massimo la vostra creativit\u00e0. Provate a creare visualizzazioni: dai diagrammi standard, a plot multi-linea con filtri complessi. L\u2019importante \u00e8 che le dashboard aiutino ad avere una chiara comprensione dei dati analizzati e dei risultati da essi estratti.<\/p>\n\n\n\n
Facci sapere se ti \u00e8 piaciuto l\u2019articolo e vuoi saperne di pi\u00f9 sugli argomenti trattati!<\/p>\n\n\n\n
Seguici per non perderti i nostri prossimi articoli su ETL e Data analytics.<\/p>\n\n\n\n
![\"serverless](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2022\/12\/image1-1024x346.png\")
<\/figure><\/div>\n\n\n![\"Athena](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2022\/12\/image3-1024x445.png\")
<\/figure><\/div>\n\n\n![\"Amazon](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2022\/12\/image2-1024x382.png\")
<\/figure><\/div>\n\n\n