Introduzione ad Apache Spark con esempi e casi d'uso

Ho sentito parlare di Spark per la prima volta alla fine del 2013, quando mi sono interessato a Scala, la lingua in cui è scritto Spark. Qualche tempo dopo, ho realizzato un divertente progetto di scienza dei dati cercando di prevedere la sopravvivenza sul Titanic. Questo si è rivelato un ottimo modo per conoscere ulteriormente i concetti e la programmazione di Spark. Lo consiglio vivamente a tutti gli aspiranti sviluppatori Spark che cercano un punto di partenza.

Oggi, Spark viene adottato dai principali attori come Amazon, eBay e Yahoo! Molte organizzazioni eseguono Spark su cluster con migliaia di nodi. Secondo le domande frequenti su Spark, il cluster più grande conosciuto ha oltre 8000 nodi. In effetti, Spark è una tecnologia che vale la pena prendere in considerazione e conoscere.

Questo articolo fornisce un'introduzione a Spark, inclusi casi d'uso ed esempi. Contiene informazioni dal sito Web di Apache Spark e dal libro Learning Spark - Lightning-Fast Big Data Analysis.

Cos'è Apache Spark? Un introduzione

Spark è un progetto Apache pubblicizzato come "cluster computing velocissimo". Ha una fiorente comunità open-source ed è il progetto Apache più attivo al momento.

Spark fornisce una piattaforma di elaborazione dati più rapida e generale. Spark ti consente di eseguire programmi fino a 100 volte più veloci in memoria o 10 volte più veloci su disco rispetto a Hadoop. L'anno scorso, Spark ha rilevato Hadoop completando il concorso Daytona GraySort da 100 TB 3 volte più velocemente con un decimo del numero di macchine ed è diventato anche il motore open source più veloce per l'ordinamento di un petabyte.

Spark consente inoltre di scrivere codice più rapidamente poiché hai a disposizione oltre 80 operatori di alto livello. Per dimostrarlo, diamo un'occhiata a "Hello World!" di BigData: l'esempio Conteggio parole. Scritto in Java per MapReduce ha circa 50 righe di codice, mentre in Spark (e Scala) puoi farlo semplicemente in questo modo:

 sparkContext.textFile("hdfs://...") .flatMap(line => line.split(" ")) .map(word => (word, 1)).reduceByKey(_ + _) .saveAsTextFile("hdfs://...")

Un altro aspetto importante quando si impara a usare Apache Spark è la shell interattiva (REPL) che fornisce immediatamente. Utilizzando REPL, è possibile testare il risultato di ogni riga di codice senza prima dover codificare ed eseguire l'intero lavoro. Il percorso verso il codice funzionante è quindi molto più breve e l'analisi dei dati ad hoc è resa possibile.

Ulteriori caratteristiche chiave di Spark includono:

• Attualmente fornisce API in Scala, Java e Python, con supporto per altri linguaggi (come R) in arrivo
• Si integra bene con l'ecosistema Hadoop e le origini dati (HDFS, Amazon S3, Hive, HBase, Cassandra, ecc.)
• Può essere eseguito su cluster gestiti da Hadoop YARN o Apache Mesos e può anche essere eseguito autonomamente

Il core Spark è completato da una serie di potenti librerie di livello superiore che possono essere utilizzate senza problemi nella stessa applicazione. Queste librerie attualmente includono SparkSQL, Spark Streaming, MLlib (per l'apprendimento automatico) e GraphX, ognuna delle quali è ulteriormente dettagliata in questo articolo. Sono attualmente in fase di sviluppo anche ulteriori librerie ed estensioni Spark.

Nucleo di scintilla

Spark Core è il motore di base per l'elaborazione dati parallela e distribuita su larga scala. È responsabile di:

• gestione della memoria e ripristino dei guasti
• pianificazione, distribuzione e monitoraggio dei lavori su un cluster
• interagire con i sistemi di archiviazione

Spark introduce il concetto di RDD (Resilient Distributed Dataset), una raccolta distribuita immutabile e tollerante agli errori su cui è possibile operare in parallelo. Un RDD può contenere qualsiasi tipo di oggetto e viene creato caricando un set di dati esterno o distribuendo una raccolta dal programma del driver.

Gli RDD supportano due tipi di operazioni:

• Le trasformazioni sono operazioni (come mappa, filtro, unione, unione e così via) che vengono eseguite su un RDD e che producono un nuovo RDD contenente il risultato.
• Le azioni sono operazioni (come riduzione, conteggio, primo e così via) che restituiscono un valore dopo aver eseguito un calcolo su un RDD.

Le trasformazioni in Spark sono "pigre", nel senso che non calcolano subito i loro risultati. Invece, "ricordano" semplicemente l'operazione da eseguire e il set di dati (es. file) su cui deve essere eseguita l'operazione. Le trasformazioni vengono effettivamente calcolate solo quando viene chiamata un'azione e il risultato viene restituito al programma del driver. Questo design consente a Spark di funzionare in modo più efficiente. Ad esempio, se un file di grandi dimensioni è stato trasformato in vari modi e passato alla prima azione, Spark elaborerà e restituirà il risultato solo per la prima riga, anziché eseguire il lavoro per l'intero file.

Per impostazione predefinita, ogni RDD trasformato può essere ricalcolato ogni volta che si esegue un'azione su di esso. Tuttavia, puoi anche rendere persistente un RDD in memoria usando il metodo persist o cache, nel qual caso Spark manterrà gli elementi presenti nel cluster per un accesso molto più rapido alla successiva query.

SparkSQL

SparkSQL è un componente Spark che supporta l'esecuzione di query sui dati tramite SQL o tramite Hive Query Language. È nato come port di Apache Hive per essere eseguito su Spark (al posto di MapReduce) ed è ora integrato con lo stack Spark. Oltre a fornire supporto per varie origini dati, consente di intrecciare query SQL con trasformazioni di codice, il che si traduce in uno strumento molto potente. Di seguito è riportato un esempio di query compatibile con Hive:

 // sc is an existing SparkContext. val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) sqlContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") sqlContext.sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src") // Queries are expressed in HiveQL sqlContext.sql("FROM src SELECT key, value").collect().foreach(println)

Scintilla in streaming

Spark Streaming supporta l'elaborazione in tempo reale dei dati in streaming, come i file di registro del server Web di produzione (ad es. Apache Flume e HDFS/S3), i social media come Twitter e varie code di messaggistica come Kafka. Sotto il cofano, Spark Streaming riceve i flussi di dati di input e divide i dati in batch. Successivamente, vengono elaborati dal motore Spark e generano il flusso finale di risultati in batch, come illustrato di seguito.

L'API Spark Streaming si avvicina molto a quella di Spark Core, rendendo facile per i programmatori lavorare nel mondo dei dati batch e di streaming.

MLlib

MLlib è una libreria di machine learning che fornisce vari algoritmi progettati per scalare su un cluster per la classificazione, la regressione, il clustering, il filtraggio collaborativo e così via (consulta l'articolo di Toptal sull'apprendimento automatico per ulteriori informazioni su questo argomento). Alcuni di questi algoritmi funzionano anche con lo streaming di dati, come la regressione lineare che utilizza i minimi quadrati ordinari o il clustering di k-medie (e altro in arrivo). Apache Mahout (una libreria di machine learning per Hadoop) si è già allontanato da MapReduce e ha unito le forze su Spark MLlib.

GraficoX

GraphX ​​è una libreria per manipolare grafici ed eseguire operazioni parallele ai grafici. Fornisce uno strumento uniforme per ETL, analisi esplorativa e calcoli di grafici iterativi. Oltre alle operazioni integrate per la manipolazione dei grafici, fornisce una libreria di algoritmi grafici comuni come PageRank.

Come utilizzare Apache Spark: caso d'uso per il rilevamento di eventi

Ora che abbiamo risposto alla domanda "Cos'è Apache Spark?", pensiamo a quale tipo di problemi o sfide potrebbe essere utilizzato in modo più efficace.

Di recente mi sono imbattuto in un articolo su un esperimento per rilevare un terremoto analizzando un flusso di Twitter. È interessante notare che è stato dimostrato che questa tecnica potrebbe informarti di un terremoto in Giappone più rapidamente dell'Agenzia meteorologica giapponese. Anche se hanno utilizzato una tecnologia diversa nel loro articolo, penso che sia un ottimo esempio per vedere come potremmo utilizzare Spark con frammenti di codice semplificati e senza il codice adesivo.

Innanzitutto, dovremmo filtrare i tweet che sembrano rilevanti come "terremoto" o "scuotimento". Potremmo facilmente utilizzare Spark Streaming per questo scopo come segue:

 TwitterUtils.createStream(...) .filter(_.getText.contains("earthquake") || _.getText.contains("shaking"))

Quindi, dovremmo eseguire un'analisi semantica sui tweet per determinare se sembrano fare riferimento a un terremoto in corso. Tweet come "Terremoto!" o "Ora sta tremando", ad esempio, sarebbero considerate corrispondenze positive, mentre tweet come "Attendere a una conferenza sul terremoto" o "Il terremoto di ieri è stato spaventoso" non lo sarebbero. Gli autori del documento hanno utilizzato una macchina vettore di supporto (SVM) per questo scopo. Faremo lo stesso qui, ma possiamo anche provare una versione in streaming. Un esempio di codice risultante da MLlib sarebbe simile al seguente:

 // We would prepare some earthquake tweet data and load it in LIBSVM format. val data = MLUtils.loadLibSVMFile(sc, "sample_earthquate_tweets.txt") // Split data into training (60%) and test (40%). val splits = data.randomSplit(Array(0.6, 0.4), seed = 11L) val training = splits(0).cache() val test = splits(1) // Run training algorithm to build the model val numIterations = 100 val model = SVMWithSGD.train(training, numIterations) // Clear the default threshold. model.clearThreshold() // Compute raw scores on the test set. val scoreAndLabels = test.map { point => val score = model.predict(point.features) (score, point.label) } // Get evaluation metrics. val metrics = new BinaryClassificationMetrics(scoreAndLabels) val auROC = metrics.areaUnderROC() println("Area under ROC = " + auROC)

Se siamo soddisfatti del tasso di previsione del modello, potremmo passare alla fase successiva e reagire ogni volta che scopriamo un terremoto. Per rilevarne uno abbiamo bisogno di un certo numero (cioè, densità) di tweet positivi in ​​una finestra temporale definita (come descritto nell'articolo). Tieni presente che, per i tweet con i servizi di localizzazione di Twitter abilitati, estrarremo anche la posizione del terremoto. Forti di questa conoscenza, potremmo utilizzare SparkSQL e interrogare una tabella Hive esistente (memorizzando gli utenti interessati a ricevere notifiche sui terremoti) per recuperare i loro indirizzi e-mail e inviare loro un'e-mail di avviso personalizzata, come segue:

 // sc is an existing SparkContext. val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) // sendEmail is a custom function sqlContext.sql("FROM earthquake_warning_users SELECT firstName, lastName, city, email") .collect().foreach(sendEmail)

Altri casi d'uso di Apache Spark

I potenziali casi d'uso di Spark vanno ben oltre il rilevamento dei terremoti, ovviamente.

Ecco un rapido (ma certamente non esaustivo!) campionamento di altri casi d'uso che richiedono di gestire la velocità, la varietà e il volume dei Big Data, per i quali Spark è così adatto:

Nell'industria dei giochi, l'elaborazione e la scoperta di schemi dalla potenziale manichetta antincendio degli eventi di gioco in tempo reale ed essere in grado di rispondere ad essi immediatamente è una capacità che potrebbe produrre un'attività redditizia, per scopi quali la fidelizzazione dei giocatori, la pubblicità mirata, l'auto -regolazione del livello di complessità, e così via.

Nel settore dell'e-commerce, le informazioni sulle transazioni in tempo reale potrebbero essere passate a un algoritmo di clustering di streaming come k-means o filtri collaborativi come ALS. I risultati potrebbero quindi essere combinati con altre fonti di dati non strutturati, come i commenti dei clienti o le recensioni dei prodotti, e utilizzati per migliorare costantemente e adattare i consigli nel tempo alle nuove tendenze.

Nel settore finanziario o della sicurezza, lo stack Spark può essere applicato a un sistema di rilevamento di frodi o intrusioni o all'autenticazione basata sul rischio. Potrebbe ottenere risultati di prim'ordine raccogliendo enormi quantità di registri archiviati, combinandoli con fonti di dati esterne come informazioni su violazioni di dati e account compromessi (vedi, ad esempio, https://haveibeenpwned.com/) e informazioni dalla connessione/ richiesta come la geolocalizzazione IP o l'ora.

Conclusione

Per riassumere, Spark aiuta a semplificare il compito impegnativo e computazionalmente intenso di elaborare elevati volumi di dati in tempo reale o archiviati, sia strutturati che non strutturati, integrando perfettamente funzionalità complesse rilevanti come l'apprendimento automatico e gli algoritmi grafici. Spark porta l'elaborazione dei Big Data alle masse. Controlla!