![\"data](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/11\/unnamed-87-1024x205.png\")
<\/figure><\/div>\n\n\n\nIn questo breve articolo andremo a presentare un’applicazione molto semplice, ma reale che abbiamo sviluppato per rappresentare la pipeline di acquisizione e analisi dei dati nel contesto di eventi e conferenze AWS e IoT. \n<\/p>\n\n\n\n
Abbiamo modificato una macchina da caff\u00e8 moderna tramite l’utilizzo di componenti elettronici personalizzati quali un Raspberry Pi Zero e una microcamera, affinch\u00e9, al premere del pulsante di erogazione, scattasse una foto alla persona di fronte e la caricasse su S3. A seguito dell\u2019upload della foto, Amazon Rekognition, triggerato da AWS Lambda, si occupa di analizzarla, individuando parametri quali:\n\n\n– esiste la persona nella foto?\n– la persona ha gli occhiali? \n– ha la barba? \n– ha la baffi? \n– sta sorridendo? <\/p>\n\n\n\n
A termine dell\u2019analisi dell\u2019immagine, se questa contiene il volto di una persona, un record viene scritto dalla funzione Lambda in un db serverless Aurora MySQL insieme all’output del modello di Machine Learning di Rekognition.\n<\/p>\n\n\n\n
Infine abbiamo sviluppato una semplicissima applicazione web collegata al database per la visualizzazione delle statistiche e una query con AWS Athena che pulisce i dati e li sposta in un bucket S3 come file parquet.\n<\/p>\n\n\n\n
Di seguito \u00e8 mostrato uno schema dell’infrastruttura proposta.<\/p>\n\n\n\n
![\"data](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-88.png\")
<\/figure><\/div>\n\n\n\nPassiamo ora a descrivere i passaggi di un comune processo di acquisizione e trasformazione dei dati e come li stiamo organizzando nella nostra applicazione.<\/p>\n\n\n\n
La fase di importazione \/ archiviazione<\/h2>\n\n\n\n
In AWS un flusso di inserimento dati molto comune consiste nell’utilizzare AWS IoT Core (Secure MQTT) o Api Gateway (API REST o Websocket) come entry point di dati, connettendolo direttamente a Kinesis Firehose (utilizzando regole IoT o Api gateway Service integrations) e infine sfruttare le potenti funzionalit\u00e0 di Firehose per il buffering dei dati, la trasformazione del buffer (funzioni AWS Lambda), la crittografia del flusso (AWS KMS), la compressione dei dati (GZIP) e la consegna dei dati a batch di messaggi compressi e crittografati automaticamente sia ad uno storage di tipo duraturo (S3) che a un data warehouse (AWS Redshift) per query analitiche complesse sull’enorme quantit\u00e0 di dati raccolti.<\/p>\n\n\n\n
Avere sempre tutti i dati acquisiti salvati in AWS S3 \u00e8 un passaggio essenziale, non solo come salvavita in caso di problemi con altri archivi pi\u00f9 caldi ma anche per creare un data lake condiviso che pu\u00f2 essere successivamente analizzato con Athena, EMR, Glue Jobs, Glue Databrew e anche strumenti esterni.<\/p>\n\n\n\n
Inoltre \u00e8 possibile utilizzare Firehose per fornire direttamente i dati ad AWS ElasticSearch per analisi in tempo reale e, se necessario, \u00e8 anche molto facile fornire i batch di dati importati ad un database relazionale (ad esempio Aurora Serverless Postgres \/ MySQL) utilizzando AWS Data Migration Service o funzioni Lambda basate su eventi. \nLa migrazione dei dati inseriti (o di un’aggregazione di essi) ad un database relazionale esistente \u00e8 spesso molto utile se si ha bisogno di arricchire un’applicazione legacy esistente che gi\u00e0 utilizza il database.<\/p>\n\n\n\n
Nel caso si decida di utilizzare le funzioni Lambda per spostare i dati importati su Aurora, metodo solitamente pi\u00f9 veloce e scalabile, \u00e8 possibile sfruttare direttamente le funzioni Lambda di trasformazione di Firehose o una funzione differente attivata ogni volta che Firehose scrive un oggetto su S3. <\/p>\n\n\n\n
![\"The](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-89-1024x576.png\")
<\/figure><\/div>\n\n\n\nUna peculiarit\u00e0 di Firehose consiste nella possibilit\u00e0 di aggiungerlo anche in un secondo momento! \nNella nostra semplice applicazione non lo utilizziamo immediatamente, pertanto le immagini e le analisi vengono salvate direttamente in S3 e AWS Aurora Serverless MySQL da Lambda Functions, nel caso in cui il flusso generato dall’applicazione dovesse crescere potremmo successivamente integrarlo senza difficolt\u00e0 alcuna!\n<\/p>\n\n\n\n
Fase di analisi<\/h2>\n\n\n\n
Una volta archiviati i dati, \u00e8 il momento di analizzarli. In questo caso le metodologie possono differire notevolmente. Gli esempi pi\u00f9 comuni vanno dalle semplici query eseguite in database relazionali, ai job analitici complessi eseguiti nei data warehouse di Redshift, oppure all’elaborazione in tempo reale utilizzando EMR o ElasticSearch.\n<\/p>\n\n\n\n
Nel nostro caso possiamo semplicemente eseguire query utilizzando il backend della nostra applicazione web e visualizzare i risultati nel browser.\n<\/p>\n\n\n\n
Tuttavia, in futuro potremmo essere interessati ad eseguire query molto pi\u00f9 avanzate sui nostri dati e magari fare qualche check sulla qualit\u00e0 del dato o training per modelli di Machine Learning. \nPer rendere possibili queste espansioni, occorre quindi spostare i dati da Aurora a S3 per analizzarli con job Glue e Databrew e, se necessario, caricarli facilmente con Apache Spark da Glue o AWS EMR. Per fare ci\u00f2 possiamo seguire diversi strade: ad esempio potremmo usare il servizio AWS DataMigration per spostare i dati su S3 come file Parquet oppure potremmo creare un Glue Job, caricare i dati usando Glue Connection da RDS con Spark e poi scriverli in S3. <\/p>\n\n\n\n
Dopo questo passaggio, sarebbe necessario eseguire un crawler Glue per creare un DataCatalog che verr\u00e0 utilizzato da Athena e Glue per query e jobs.<\/p>\n\n\n\n
Qui tuttavia mostreremo un percorso diverso e talvolta molto pi\u00f9 flessibile per esportare i dati in modo pulito e catalogarli da un database relazionale: <\/span>Athena custom data source.<\/b><\/p>\n\n\n\n Per impostazione predefinita, Athena viene fornito con l\u2019integrazione S3 – Glue Data Catalog, ma di recente AWS ha dato la possibilit\u00e0 di aggiungere un’origine dati personalizzata, ad esempio database connessi tramite JDBC, AWS CloudWatch o l\u2019esecuzione di query su S3 ma utilizzando un metastore Apache Hive personalizzato. Nel nostro caso siamo interessati a connetterci a MySQL Aurora Serverless. Per fare ci\u00f2 dobbiamo andare su Athena Home, configurare un workgroup denominato AmazonAthenaPreviewFunctionality e quindi aggiungere una path S3 di output delle query di Athena:<\/p>\n\n\n\n Dopodich\u00e9 possiamo tornare alla home di Athena e selezionare Connect Data Source:<\/p>\n\n\n\n Ci viene presentata una pagina web dove dobbiamo selezionare il tipo di sorgente dati: optiamo per Query a data source (beta) e MySQL:\n<\/p>\n\n\n\n A questo punto ci viene chiesto di inserire il nome e la descrizione del nuovo catalogo e di selezionare o creare una funzione Lambda per gestire la connessione. Impostiamo il nome desiderato e facciamo clic su Configura nuova funzione AWS Lambda.<\/p>\n\n\n\n Viene presentata questa pagina in cui bisogna inserire l’uri di connessione JDBC per Aurora e selezionare la subnet e il security group per la funzione Lambda che Athena utilizzer\u00e0 per stabilire la connessione JDBC. Vanno scelti accuratamente, altrimenti la \u200b\u200bLambda non raggiunger\u00e0 l’istanza Aurora!<\/span><\/p>\n\n\n\n Il prefisso del Secret viene utilizzato per memorizzare le credenziali del database in AWS Secret Manager, questo \u00e8 essenziale per l’ambiente di produzione e se lo si lascia vuoto, l’integrazione non verr\u00e0 creata. \nDopo aver selezionato deploy e selezionato la Lambda appena creata nella dashboard di Athena, verr\u00e0 creato un nuovo catalogo diverso dallo standard AwsGlueCatalog:<\/p>\n\n\n\n Tuttavia all’inizio Databeses e tabelle non appariranno. Controllando i log della Lambda su CloudWatch si trover\u00e0 un errore del tipo:<\/p>\n\n\n\n Impostiamo quindi la variabile di environment richiesta per la funzione Lambda utilizzando la stessa stringa di connessione JDBC usata come stringa DefaultConnection nel passaggio precedente. Dopodich\u00e9 la connessione funzioner\u00e0 e si potr\u00e0 interrogare il DB direttamente da Athena! Ottimo!<\/p>\n\n\n\n Ad uno sguardo pi\u00f9 attento, per\u00f2, notiamo immediatamente che qualcosa \u00e8 in conflitto con i dati: ecco una schermata di ci\u00f2 che possiamo leggere direttamente da MySQL:<\/p>\n\n\n\n Come si pu\u00f2 vedere Athena \u00e8 abbastanza intelligente da convertire i dati tinyint(1) in bool ma non riesce a leggere le colonne datetime da mysql. Ci\u00f2 \u00e8 dovuto ad un problema molto noto con il connettore jdbc e la soluzione pi\u00f9 semplice \u00e8 creare un nuovo campo dove datetime \u00e8 una stringa in formato datetime Java:<\/p>\n\n\n\n A questo punto Athena potr\u00e0 leggere il nuovo campo.\nEd ora siamo pronti per un bellissimo trucco: <\/span>\n<\/span>andiamo nella dashboard di Glue e creiamo un nuovo Database: un database \u00e8 solo un contenitore logico per metadati, si pu\u00f2 scegliere il nome che si preferisce.<\/span><\/p>\n\n\n\n A questo punto possiamo tornare ad Athena ed eseguire una query come questa:<\/p>\n\n\n\n Ci\u00f2 creer\u00e0 una nuova tabella nel database che abbiamo appena aggiunto al nostro catalogo di dati Glue e salver\u00e0 tutti i dati in S3 come file GZIP Parquet. Inoltre \u00e8 anche possibile, se lo si desidera, cambiare la compressione (es. Snappy o BZIP).<\/p>\n\n\n\n Oltre ad esportare i dati come Parquet, la query andr\u00e0 ad eliminare quelli con la formattazione errata per l\u2019url di S3.<\/p>\n\n\n\n Abbiamo quindi un modo super veloce per esportare il nostro db in S3 come parquet e creare automaticamente il catalogo Glue.\n<\/p>\n\n\n\n Diventa quindi semplice visualizzare questo nuovo catalogo in AWS Glue Databrew: andiamo alla dashboard di Databrew e creiamo un nuovo progetto <\/p>\n\n\n\n e un nuovo set di dati nella sezione aggiungi dataset.<\/p>\n\n\n\n In caso di errore sar\u00e0 necessario rinominare il file s3 come .parquet e scansionare nuovamente la tabella con i crawler Glue.\n<\/p>\n\n\n\n Et voil\u00e0 una bellissima visualizzazione dei dati del nostro dataset completa di statistiche delle colonne!<\/p>\n\n\n\n In questo articolo abbiamo descritto un’applicazione IoT molto semplice che utilizza Amazon Rekognition e Amazon Aurora. Abbiamo spiegato come pu\u00f2 essere migliorata sfruttando Firehose e infine abbiamo utilizzato Athena per trasformare e pulire i dati raccolti.<\/p>\n\n\n\n Abbiamo visto anche come salvarli molto facilmente come parquet e come possono essere analizzati con Glue Databrew, Athena e altri strumenti AWS come EMR.<\/span><\/p>\n\n\n\n Avete mai provato configurazioni simili per i vostri processi di Data Analysis?<\/p>\n\n\n\n Fateci sapere! Saremo felici di offrirvi un caff\u00e8…connesso \ud83d\ude00 \n<\/p>\n\n\n\n Per oggi \u00e8 tutto. <\/p>\n\n\n\n Continuate a seguirci: ci vediamo tra 14 giorni qui su #Proud2beCloud!<\/strong><\/p>\n","protected":false},"excerpt":{"rendered":" Con l\u2019avvento dell\u2019Internet of Things, il numero di device connessi sta aumentando in modo esponenziale, cos\u00ec come la quantit\u00e0 di […]<\/p>\n","protected":false},"author":9,"featured_media":2036,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[477,474],"tags":[414,261,426,251,395,422,410,418,420,424,412],"yoast_head":"\n![\"configure](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-90-1024x236.png\")
<\/figure><\/div>\n\n\n\n![\"Connect](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-91.png\")
<\/figure><\/div>\n\n\n\n![\"Query](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-92-1024x542.png\")
<\/figure><\/div>\n\n\n\n![\"Configure](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-93-1024x456.png\")
<\/figure><\/div>\n\n\n\n![\"JDBC](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-94-1024x1014.png\")
<\/figure><\/div>\n\n\n\n![\"\"](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-95.png\")
<\/figure><\/div>\n\n\n\nCatalog is not supported in multiplexer. After registering the catalog in Athena, must set 'iotarticolo_connection_string' environment variable in Lambda. See JDBC connector README for further details.: java.lang.RuntimeException<\/pre>\n\n\n\n
![\"DB](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-96-1024x365.png\")
<\/figure><\/div>\n\n\n\n![\"Athena](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-97-1024x142.png\")
<\/figure><\/div>\n\n\n\nUPDATE coffees SET coffees.coffee_hour_str=DATE_FORMAT(coffee_hour, '%Y-%m-%d %H:%i:%s');\nALTER TABLE coffees ADD COLUMN coffee_hour_str VARCHAR(255) AFTER coffee_hour;\n<\/pre>\n\n\n\n
![\"create](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-98-1024x555.png\")
<\/figure><\/div>\n\n\n\nCREATE table iotarticologlue.coffees<\/span>\n\nWITH (<\/span>\n\n format='PARQUET', external_location='s3:\/\/besharp-athena\/coffees_parquet', parquet_compression='GZIP'<\/span>\n\n) AS SELECT photo_url,smile,beard,mustache,glasses,coffee_hour_str FROM \"iotarticolo\".\"iot\".\"coffees\" <\/span>\n\nWHERE photo_url LIKE 'https:\/\/%';<\/span><\/pre>\n\n\n\n![\"run](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-99-1024x166.png\")
<\/figure><\/div>\n\n\n\n![\"a](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-100-1024x576.png\")
<\/figure><\/div>\n\n\n\n![\"AWS](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-2020-12-11T175425.819-1024x182.png\")
<\/figure><\/div>\n\n\n\n![\"AWS](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-2020-12-11T175558.801-1024x629.png\")
<\/figure><\/div>\n\n\n\n![\"data](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2020\/12\/unnamed-2020-12-11T175732.127-1024x470.png\")
<\/figure><\/div>\n\n\n\nConclusioni<\/h2>\n\n\n\n