Spark SQL - Guida rapida

Le industrie utilizzano ampiamente Hadoop per analizzare i propri set di dati. Il motivo è che il framework Hadoop si basa su un semplice modello di programmazione (MapReduce) e consente una soluzione di elaborazione scalabile, flessibile, a tolleranza di errore e conveniente. In questo caso, la preoccupazione principale è mantenere la velocità nell'elaborazione di grandi set di dati in termini di tempo di attesa tra le query e tempo di attesa per eseguire il programma.

Spark è stato introdotto da Apache Software Foundation per accelerare il processo del software di calcolo computazionale Hadoop.

Contro una credenza comune, Spark is not a modified version of Hadoope non dipende, in realtà, da Hadoop perché dispone di una propria gestione del cluster. Hadoop è solo uno dei modi per implementare Spark.

Spark utilizza Hadoop in due modi: uno è storage e il secondo è processing. Poiché Spark ha il proprio calcolo di gestione del cluster, utilizza Hadoop solo a scopo di archiviazione.

Apache Spark

Apache Spark è una tecnologia di elaborazione cluster velocissima, progettata per un calcolo veloce. È basato su Hadoop MapReduce ed estende il modello MapReduce per utilizzarlo in modo efficiente per più tipi di calcoli, che includono query interattive e elaborazione del flusso. La caratteristica principale di Spark è la suain-memory cluster computing che aumenta la velocità di elaborazione di un'applicazione.

Spark è progettato per coprire un'ampia gamma di carichi di lavoro come applicazioni batch, algoritmi iterativi, query interattive e streaming. Oltre a supportare tutti questi carichi di lavoro in un rispettivo sistema, riduce l'onere di gestione del mantenimento di strumenti separati.

Evoluzione di Apache Spark

Spark è uno dei sottoprogetti di Hadoop sviluppati nel 2009 nell'AMPLab di UC Berkeley da Matei Zaharia. Era Open Sourced nel 2010 con una licenza BSD. È stato donato alla Fondazione software Apache nel 2013 e ora Apache Spark è diventato un progetto Apache di primo livello da febbraio 2014.

Caratteristiche di Apache Spark

Apache Spark ha le seguenti caratteristiche.

  • Speed- Spark aiuta a eseguire un'applicazione nel cluster Hadoop, fino a 100 volte più veloce in memoria e 10 volte più veloce durante l'esecuzione su disco. Ciò è possibile riducendo il numero di operazioni di lettura / scrittura su disco. Memorizza in memoria i dati di elaborazione intermedia.

  • Supports multiple languages- Spark fornisce API integrate in Java, Scala o Python. Pertanto, puoi scrivere applicazioni in diverse lingue. Spark presenta 80 operatori di alto livello per le query interattive.

  • Advanced Analytics- Spark non supporta solo "Mappa" e "Riduci". Supporta anche query SQL, dati in streaming, machine learning (ML) e algoritmi di grafici.

Spark costruito su Hadoop

Il diagramma seguente mostra tre modi per creare Spark con i componenti Hadoop.

Esistono tre modi per la distribuzione di Spark, come spiegato di seguito.

  • Standalone- La distribuzione autonoma di Spark significa che Spark occupa il posto sopra HDFS (Hadoop Distributed File System) e lo spazio viene allocato esplicitamente per HDFS. Qui, Spark e MapReduce verranno eseguiti fianco a fianco per coprire tutti i processi Spark sul cluster.

  • Hadoop Yarn- La distribuzione di Hadoop Yarn significa, semplicemente, spark funziona su Yarn senza alcuna preinstallazione o accesso root richiesto. Aiuta a integrare Spark nell'ecosistema Hadoop o nello stack Hadoop. Consente ad altri componenti di funzionare in cima allo stack.

  • Spark in MapReduce (SIMR)- Spark in MapReduce viene utilizzato per avviare il lavoro Spark oltre alla distribuzione autonoma. Con SIMR, l'utente può avviare Spark e utilizza la sua shell senza alcun accesso amministrativo.

Componenti di Spark

La figura seguente mostra i diversi componenti di Spark.

Apache Spark Core

Spark Core è il motore di esecuzione generale sottostante per la piattaforma Spark su cui si basano tutte le altre funzionalità. Fornisce elaborazione in memoria e set di dati di riferimento in sistemi di archiviazione esterni.

Spark SQL

Spark SQL è un componente in cima a Spark Core che introduce una nuova astrazione dei dati chiamata SchemaRDD, che fornisce supporto per dati strutturati e semi-strutturati.

Spark Streaming

Spark Streaming sfrutta la capacità di pianificazione rapida di Spark Core per eseguire analisi di streaming. Acquisisce i dati in mini-batch ed esegue trasformazioni RDD (Resilient Distributed Datasets) su questi mini-batch di dati.

MLlib (libreria di machine learning)

MLlib è un framework di machine learning distribuito sopra Spark a causa dell'architettura Spark basata sulla memoria distribuita. Secondo i benchmark, è fatto dagli sviluppatori MLlib contro le implementazioni ALS (Alternating Least Squares). Spark MLlib è nove volte più veloce della versione basata su disco Hadoop diApache Mahout (prima che Mahout acquisisse un'interfaccia Spark).

GraphX

GraphX ​​è un framework di elaborazione di grafici distribuito su Spark. Fornisce un'API per esprimere il calcolo del grafico in grado di modellare i grafici definiti dall'utente utilizzando l'API di astrazione Pregel. Fornisce inoltre un runtime ottimizzato per questa astrazione.

Set di dati distribuiti resilienti

Resilient Distributed Datasets (RDD) è una struttura dati fondamentale di Spark. È una raccolta distribuita immutabile di oggetti. Ogni set di dati in RDD è suddiviso in partizioni logiche, che possono essere calcolate su diversi nodi del cluster. Gli RDD possono contenere qualsiasi tipo di oggetti Python, Java o Scala, comprese le classi definite dall'utente.

Formalmente, un RDD è una raccolta di record partizionata di sola lettura. Gli RDD possono essere creati tramite operazioni deterministiche sui dati su una memoria stabile o su altri RDD. RDD è una raccolta di elementi a tolleranza di errore su cui è possibile operare in parallelo.

Esistono due modi per creare RDD: parallelizing una raccolta esistente nel programma del driver o referencing a dataset in un sistema di archiviazione esterno, come un file system condiviso, HDFS, HBase o qualsiasi origine dati che offra un formato di input Hadoop.

Spark fa uso del concetto di RDD per ottenere operazioni MapReduce più veloci ed efficienti. Discutiamo prima di come avvengono le operazioni di MapReduce e perché non sono così efficienti.

La condivisione dei dati è lenta in MapReduce

MapReduce è ampiamente adottato per l'elaborazione e la generazione di grandi set di dati con un algoritmo parallelo e distribuito su un cluster. Consente agli utenti di scrivere calcoli paralleli, utilizzando una serie di operatori di alto livello, senza doversi preoccupare della distribuzione del lavoro e della tolleranza ai guasti.

Sfortunatamente, nella maggior parte dei framework attuali, l'unico modo per riutilizzare i dati tra i calcoli (Es: tra due lavori MapReduce) è scriverli su un sistema di archiviazione stabile esterno (Es: HDFS). Sebbene questo framework fornisca numerose astrazioni per accedere alle risorse di calcolo di un cluster, gli utenti vogliono ancora di più.

Tutti e due Iterative e Interactivele applicazioni richiedono una condivisione dei dati più rapida tra lavori paralleli. La condivisione dei dati è lenta in MapReduce a causa direplication, serialization, e disk IO. Per quanto riguarda il sistema di archiviazione, la maggior parte delle applicazioni Hadoop trascorrono più del 90% del tempo in operazioni di lettura-scrittura HDFS.

Operazioni iterative su MapReduce

Riutilizza i risultati intermedi in più calcoli in applicazioni in più fasi. La figura seguente spiega come funziona il framework corrente, mentre si eseguono le operazioni iterative su MapReduce. Ciò comporta notevoli sovraccarichi a causa della replica dei dati, dell'I / O del disco e della serializzazione, che rallentano il sistema.

Operazioni interattive su MapReduce

L'utente esegue query ad-hoc sullo stesso sottoinsieme di dati. Ogni query eseguirà l'I / O del disco sulla memoria stabile, che può dominare il tempo di esecuzione dell'applicazione.

La figura seguente spiega come funziona il framework corrente durante l'esecuzione delle query interattive su MapReduce.

Condivisione dei dati utilizzando Spark RDD

La condivisione dei dati è lenta in MapReduce a causa di replication, serialization, e disk IO. Nella maggior parte delle applicazioni Hadoop, trascorrono più del 90% del tempo in operazioni di lettura e scrittura HDFS.

Riconoscendo questo problema, i ricercatori hanno sviluppato un framework specializzato chiamato Apache Spark. L'idea chiave della scintilla èResiliente Ddistribuito Datasets (RDD); supporta il calcolo dell'elaborazione in memoria. Ciò significa che memorizza lo stato della memoria come un oggetto tra i lavori e l'oggetto è condivisibile tra questi lavori. La condivisione dei dati in memoria è da 10 a 100 volte più veloce della rete e del disco.

Proviamo ora a scoprire come avvengono le operazioni iterative e interattive in Spark RDD.

Operazioni iterative su Spark RDD

L'illustrazione riportata di seguito mostra le operazioni iterative su Spark RDD. Memorizzerà i risultati intermedi in una memoria distribuita invece che nell'archiviazione stabile (disco) e renderà il sistema più veloce.

Note - Se la memoria distribuita (RAM) è sufficiente per memorizzare i risultati intermedi (stato del lavoro), memorizzerà quei risultati sul disco

Operazioni interattive su Spark RDD

Questa illustrazione mostra le operazioni interattive su Spark RDD. Se diverse query vengono eseguite ripetutamente sullo stesso set di dati, questi particolari dati possono essere conservati in memoria per tempi di esecuzione migliori.

Per impostazione predefinita, ogni RDD trasformato può essere ricalcolato ogni volta che si esegue un'azione su di esso. Tuttavia, potresti anchepersistun RDD in memoria, nel qual caso Spark manterrà gli elementi nel cluster per un accesso molto più rapido, la prossima volta che lo interrogherai. È inoltre disponibile il supporto per la persistenza di RDD su disco o la replica su più nodi.

Spark è il sottoprogetto di Hadoop. Pertanto, è meglio installare Spark in un sistema basato su Linux. I seguenti passaggi mostrano come installare Apache Spark.

Passaggio 1: verifica dell'installazione di Java

L'installazione di Java è una delle cose obbligatorie nell'installazione di Spark. Prova il seguente comando per verificare la versione JAVA.

$java -version

Se Java è già installato sul tuo sistema, puoi vedere la seguente risposta:

java version "1.7.0_71"
Java(TM) SE Runtime Environment (build 1.7.0_71-b13)
Java HotSpot(TM) Client VM (build 25.0-b02, mixed mode)

Se non hai Java installato sul tuo sistema, installa Java prima di procedere al passaggio successivo.

Passaggio 2: verifica dell'installazione di Scala

Dovresti usare il linguaggio Scala per implementare Spark. Quindi verifichiamo l'installazione di Scala usando il seguente comando.

$scala -version

Se Scala è già installato sul tuo sistema, vedrai la seguente risposta:

Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

Nel caso in cui non hai Scala installato sul tuo sistema, procedi al passaggio successivo per l'installazione di Scala.

Passaggio 3: download di Scala

Scarica l'ultima versione di Scala visitando il seguente link Download Scala . Per questo tutorial, stiamo usando la versione scala-2.11.6. Dopo il download, troverai il file tar di Scala nella cartella di download.

Passaggio 4: installazione di Scala

Seguire i passaggi indicati di seguito per l'installazione di Scala.

Estrai il file tar di Scala

Digita il seguente comando per estrarre il file tar Scala.

$ tar xvf scala-2.11.6.tgz

Spostare i file del software Scala

Utilizzare i seguenti comandi per spostare i file del software Scala nella rispettiva directory (/usr/local/scala).

$ su –
Password:
# cd /home/Hadoop/Downloads/
# mv scala-2.11.6 /usr/local/scala
# exit

Imposta PATH per Scala

Utilizzare il seguente comando per impostare PATH per Scala.

$ export PATH = $PATH:/usr/local/scala/bin

Verifica dell'installazione di Scala

Dopo l'installazione, è meglio verificarlo. Utilizzare il seguente comando per verificare l'installazione di Scala.

$scala -version

Se Scala è già installato sul tuo sistema, vedrai la seguente risposta:

Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

Passaggio 5: download di Apache Spark

Scarica l'ultima versione di Spark visitando il seguente link Download Spark . Per questo tutorial, stiamo usandospark-1.3.1-bin-hadoop2.6versione. Dopo averlo scaricato, troverai il file tar Spark nella cartella di download.

Passaggio 6: installazione di Spark

Segui i passaggi indicati di seguito per l'installazione di Spark.

Estrazione del catrame di scintilla

Il seguente comando per estrarre il file spark tar.

$ tar xvf spark-1.3.1-bin-hadoop2.6.tgz

Spostamento dei file del software Spark

I seguenti comandi per spostare i file del software Spark nella rispettiva directory (/usr/local/spark).

$ su –
Password:
# cd /home/Hadoop/Downloads/
# mv spark-1.3.1-bin-hadoop2.6 /usr/local/spark
# exit

Configurazione dell'ambiente per Spark

Aggiungi la riga seguente a ~/.bashrcfile. Significa aggiungere la posizione, in cui si trova il file del software spark alla variabile PATH.

export PATH = $PATH:/usr/local/spark/bin

Utilizzare il seguente comando per reperire il file ~ / .bashrc.

$ source ~/.bashrc

Passaggio 7: verifica dell'installazione di Spark

Scrivi il seguente comando per aprire la shell Spark.

$spark-shell

Se spark è installato correttamente, troverai il seguente output.

Spark assembly has been built with Hive, including Datanucleus jars on classpath
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
15/06/04 15:25:22 INFO SecurityManager: Changing view acls to: hadoop
15/06/04 15:25:22 INFO SecurityManager: Changing modify acls to: hadoop
disabled; ui acls disabled; users with view permissions: Set(hadoop); users with modify permissions: Set(hadoop)
15/06/04 15:25:22 INFO HttpServer: Starting HTTP Server
15/06/04 15:25:23 INFO Utils: Successfully started service 'HTTP class server' on port 43292.
Welcome to
    ____             __
   / __/__ ___ _____/ /__
   _\ \/ _ \/ _ `/ __/ '_/
   /___/ .__/\_,_/_/ /_/\_\ version 1.4.0
      /_/
Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71)
Type in expressions to have them evaluated.
Spark context available as sc
scala>

Spark introduce un modulo di programmazione per l'elaborazione dei dati strutturati chiamato Spark SQL. Fornisce un'astrazione di programmazione chiamata DataFrame e può fungere da motore di query SQL distribuito.

Caratteristiche di Spark SQL

Le seguenti sono le funzionalità di Spark SQL:

  • Integrated- Combina perfettamente le query SQL con i programmi Spark. Spark SQL ti consente di eseguire query sui dati strutturati come un set di dati distribuito (RDD) in Spark, con API integrate in Python, Scala e Java. Questa stretta integrazione semplifica l'esecuzione di query SQL insieme a complessi algoritmi analitici.

  • Unified Data Access- Carica ed esegui query sui dati da una varietà di origini. Gli Schema-RDD forniscono un'unica interfaccia per lavorare in modo efficiente con i dati strutturati, comprese le tabelle Apache Hive, i file parquet e i file JSON.

  • Hive Compatibility- Esegui query Hive non modificate sui magazzini esistenti. Spark SQL riutilizza il frontend Hive e MetaStore, offrendoti piena compatibilità con i dati, le query e le UDF esistenti di Hive. Installalo semplicemente insieme a Hive.

  • Standard Connectivity- Connessione tramite JDBC o ODBC. Spark SQL include una modalità server con connettività JDBC e ODBC standard del settore.

  • Scalability- Utilizza lo stesso motore per query sia interattive che lunghe. Spark SQL sfrutta il modello RDD per supportare la tolleranza agli errori di query intermedie, consentendo la scalabilità anche a lavori di grandi dimensioni. Non preoccuparti di utilizzare un motore diverso per i dati storici.

Spark SQL Architecture

La figura seguente spiega l'architettura di Spark SQL:

Questa architettura contiene tre livelli, ovvero Language API, Schema RDD e Data Sources.

  • Language API- Spark è compatibile con diversi linguaggi e Spark SQL. È anche supportato da questi linguaggi: API (python, scala, java, HiveQL).

  • Schema RDD- Spark Core è progettato con una struttura dati speciale chiamata RDD. In genere, Spark SQL funziona su schemi, tabelle e record. Pertanto, possiamo usare lo Schema RDD come tabella temporanea. Possiamo chiamare questo schema RDD come data frame.

  • Data Sources- Di solito l'origine dati per Spark-core è un file di testo, un file Avro, ecc. Tuttavia, le origini dati per Spark SQL sono diverse. Quelli sono file Parquet, documento JSON, tabelle HIVE e database Cassandra.

Discuteremo di più su questi nei capitoli successivi.

Un DataFrame è una raccolta distribuita di dati, organizzata in colonne denominate. Concettualmente, è equivalente a tabelle relazionali con buone tecniche di ottimizzazione.

Un DataFrame può essere costruito da una matrice di origini diverse come tabelle Hive, file di dati strutturati, database esterni o RDD esistenti. Questa API è stata progettata per applicazioni moderne di Big Data e data science prendendo ispirazioneDataFrame in R Programming e Pandas in Python.

Caratteristiche di DataFrame

Ecco una serie di alcune caratteristiche caratteristiche di DataFrame:

  • Capacità di elaborare i dati nella dimensione da Kilobyte a Petabyte su un cluster a nodo singolo in cluster di grandi dimensioni.

  • Supporta diversi formati di dati (Avro, csv, ricerca elastica e Cassandra) e sistemi di archiviazione (HDFS, tabelle HIVE, mysql, ecc.).

  • Ottimizzazione dello stato dell'arte e generazione di codice tramite l'ottimizzatore Spark SQL Catalyst (framework di trasformazione dell'albero).

  • Può essere facilmente integrato con tutti gli strumenti e framework di Big Data tramite Spark-Core.

  • Fornisce API per la programmazione Python, Java, Scala e R.

SQLContext

SQLContext è una classe e viene utilizzata per inizializzare le funzionalità di Spark SQL. SparkContext classe oggetto (sc) è necessario per inizializzare l'oggetto classe SQLContext.

Il comando seguente viene utilizzato per inizializzare SparkContext tramite spark-shell.

$ spark-shell

Per impostazione predefinita, l'oggetto SparkContext viene inizializzato con il nome sc quando parte la scintilla.

Utilizzare il comando seguente per creare SQLContext.

scala> val sqlcontext = new org.apache.spark.sql.SQLContext(sc)

Esempio

Consideriamo un esempio di record dei dipendenti in un file JSON denominato employee.json. Utilizzare i seguenti comandi per creare un DataFrame (df) e leggere un documento JSON denominatoemployee.json con il seguente contenuto.

employee.json - Posiziona questo file nella directory in cui si trova il file scala> il puntatore si trova.

{
   {"id" : "1201", "name" : "satish", "age" : "25"}
   {"id" : "1202", "name" : "krishna", "age" : "28"}
   {"id" : "1203", "name" : "amith", "age" : "39"}
   {"id" : "1204", "name" : "javed", "age" : "23"}
   {"id" : "1205", "name" : "prudvi", "age" : "23"}
}

Operazioni DataFrame

DataFrame fornisce un linguaggio specifico del dominio per la manipolazione dei dati strutturati. Qui, includiamo alcuni esempi di base di elaborazione di dati strutturati utilizzando DataFrame.

Seguire i passaggi indicati di seguito per eseguire le operazioni DataFrame -

Leggi il documento JSON

Innanzitutto, dobbiamo leggere il documento JSON. Sulla base di ciò, genera un DataFrame denominato (dfs).

Utilizza il seguente comando per leggere il documento JSON denominato employee.json. I dati vengono visualizzati come una tabella con i campi: id, nome ed età.

scala> val dfs = sqlContext.read.json("employee.json")

Output - I nomi dei campi vengono presi automaticamente da employee.json.

dfs: org.apache.spark.sql.DataFrame = [age: string, id: string, name: string]

Mostra i dati

Se vuoi vedere i dati nel DataFrame, usa il seguente comando.

scala> dfs.show()

Output - Puoi vedere i dati dei dipendenti in formato tabulare.

<console>:22, took 0.052610 s
+----+------+--------+
|age | id   |  name  |
+----+------+--------+
| 25 | 1201 | satish |
| 28 | 1202 | krishna|
| 39 | 1203 | amith  |
| 23 | 1204 | javed  |
| 23 | 1205 | prudvi |
+----+------+--------+

Usa il metodo printSchema

Se vuoi vedere la struttura (schema) del DataFrame, usa il seguente comando.

scala> dfs.printSchema()

Output

root
   |-- age: string (nullable = true)
   |-- id: string (nullable = true)
   |-- name: string (nullable = true)

Usa metodo di selezione

Usa il seguente comando per recuperare name-colonna tra tre colonne dal DataFrame.

scala> dfs.select("name").show()

Output - Puoi vedere i valori di name colonna.

<console>:22, took 0.044023 s
+--------+
|  name  |
+--------+
| satish |
| krishna|
| amith  |
| javed  |
| prudvi |
+--------+

Usa filtro età

Utilizzare il seguente comando per trovare i dipendenti la cui età è maggiore di 23 (età> 23).

scala> dfs.filter(dfs("age") > 23).show()

Output

<console>:22, took 0.078670 s
+----+------+--------+
|age | id   | name   |
+----+------+--------+
| 25 | 1201 | satish |
| 28 | 1202 | krishna|
| 39 | 1203 | amith  |
+----+------+--------+

Usa il metodo groupBy

Utilizzare il seguente comando per contare il numero di dipendenti che hanno la stessa età.

scala> dfs.groupBy("age").count().show()

Output - due dipendenti hanno 23 anni.

<console>:22, took 5.196091 s
+----+-----+
|age |count|
+----+-----+
| 23 |  2  |
| 25 |  1  |
| 28 |  1  |
| 39 |  1  |
+----+-----+

Esecuzione di query SQL a livello di programmazione

Un SQLContext consente alle applicazioni di eseguire query SQL a livello di codice durante l'esecuzione di funzioni SQL e restituisce il risultato come DataFrame.

Generalmente, in background, SparkSQL supporta due diversi metodi per convertire RDD esistenti in DataFrame:

Sr. No Metodi e descrizione
1 Deduzione dello schema utilizzando la riflessione

Questo metodo utilizza la riflessione per generare lo schema di un RDD che contiene tipi specifici di oggetti.
2 Specifica dello schema a livello di codice

Il secondo metodo per creare DataFrame è tramite l'interfaccia programmatica che consente di costruire uno schema e quindi applicarlo a un RDD esistente.

Un'interfaccia DataFrame consente a diverse DataSource di lavorare su Spark SQL. È un tavolo temporaneo e può essere utilizzato come un normale RDD. La registrazione di un DataFrame come tabella consente di eseguire query SQL sui suoi dati.

In questo capitolo, descriveremo i metodi generali per caricare e salvare i dati utilizzando diverse origini dati Spark. Successivamente, discuteremo in dettaglio le opzioni specifiche disponibili per le origini dati integrate.

Esistono diversi tipi di origini dati disponibili in SparkSQL, alcune delle quali sono elencate di seguito:

Sr. No Origine dei dati
1 Set di dati JSON

Spark SQL può acquisire automaticamente lo schema di un set di dati JSON e caricarlo come DataFrame.
2 Tabelle dell'alveare

Hive viene fornito in bundle con la libreria Spark come HiveContext, che eredita da SQLContext.
3 File per parquet

Il parquet è un formato colonnare, supportato da molti sistemi di elaborazione dati.