Spark SQL - Guía rápida

Las industrias están usando Hadoop de manera extensiva para analizar sus conjuntos de datos. La razón es que el marco Hadoop se basa en un modelo de programación simple (MapReduce) y permite una solución informática que es escalable, flexible, tolerante a fallas y rentable. Aquí, la principal preocupación es mantener la velocidad en el procesamiento de grandes conjuntos de datos en términos de tiempo de espera entre consultas y tiempo de espera para ejecutar el programa.

Spark fue presentado por Apache Software Foundation para acelerar el proceso de software de computación computacional Hadoop.

En contra de una creencia común, Spark is not a modified version of Hadoopy, en realidad, no depende de Hadoop porque tiene su propia gestión de clústeres. Hadoop es solo una de las formas de implementar Spark.

Spark usa Hadoop de dos formas: una es storage y el segundo es processing. Dado que Spark tiene su propio cálculo de administración de clústeres, usa Hadoop solo con fines de almacenamiento.

Apache Spark

Apache Spark es una tecnología de computación en clúster ultrarrápida, diseñada para una computación rápida. Se basa en Hadoop MapReduce y extiende el modelo MapReduce para usarlo de manera eficiente para más tipos de cálculos, que incluyen consultas interactivas y procesamiento de flujo. La característica principal de Spark es suin-memory cluster computing que aumenta la velocidad de procesamiento de una aplicación.

Spark está diseñado para cubrir una amplia gama de cargas de trabajo, como aplicaciones por lotes, algoritmos iterativos, consultas interactivas y transmisión. Además de soportar todas estas cargas de trabajo en un sistema respectivo, reduce la carga de gestión de mantener herramientas separadas.

Evolución de Apache Spark

Spark es uno de los subproyectos de Hadoop desarrollado en 2009 en el AMPLab de UC Berkeley por Matei Zaharia. Fue de código abierto en 2010 bajo una licencia BSD. Fue donado a la fundación de software Apache en 2013, y ahora Apache Spark se ha convertido en un proyecto de Apache de alto nivel desde febrero de 2014.

Características de Apache Spark

Apache Spark tiene las siguientes características.

  • Speed- Spark ayuda a ejecutar una aplicación en el clúster de Hadoop, hasta 100 veces más rápido en memoria y 10 veces más rápido cuando se ejecuta en disco. Esto es posible reduciendo el número de operaciones de lectura / escritura en el disco. Almacena los datos de procesamiento intermedio en la memoria.

  • Supports multiple languages- Spark proporciona API integradas en Java, Scala o Python. Por lo tanto, puede escribir aplicaciones en diferentes idiomas. Spark presenta 80 operadores de alto nivel para consultas interactivas.

  • Advanced Analytics- Spark no solo es compatible con 'Map' y 'reduce'. También admite consultas SQL, transmisión de datos, aprendizaje automático (ML) y algoritmos de gráficos.

Spark construido en Hadoop

El siguiente diagrama muestra tres formas de construir Spark con componentes de Hadoop.

Hay tres formas de implementación de Spark, como se explica a continuación.

  • Standalone- La implementación independiente de Spark significa que Spark ocupa el lugar en la parte superior de HDFS (sistema de archivos distribuido de Hadoop) y el espacio se asigna para HDFS, explícitamente. Aquí, Spark y MapReduce se ejecutarán uno al lado del otro para cubrir todos los trabajos de Spark en el clúster.

  • Hadoop Yarn- La implementación de Hadoop Yarn significa, simplemente, Spark se ejecuta en Yarn sin necesidad de preinstalación o acceso de root. Ayuda a integrar Spark en el ecosistema de Hadoop o la pila de Hadoop. Permite que otros componentes se ejecuten en la parte superior de la pila.

  • Spark in MapReduce (SIMR)- Spark en MapReduce se usa para iniciar el trabajo de Spark además de la implementación independiente. Con SIMR, el usuario puede iniciar Spark y usar su shell sin ningún acceso administrativo.

Componentes de Spark

La siguiente ilustración muestra los diferentes componentes de Spark.

Apache Spark Core

Spark Core es el motor de ejecución general subyacente para la plataforma Spark sobre el que se basan todas las demás funciones. Proporciona cómputo en memoria y conjuntos de datos de referencia en sistemas de almacenamiento externos.

Spark SQL

Spark SQL es un componente sobre Spark Core que presenta una nueva abstracción de datos llamada SchemaRDD, que brinda soporte para datos estructurados y semiestructurados.

Spark Streaming

Spark Streaming aprovecha la capacidad de programación rápida de Spark Core para realizar análisis de transmisión. Ingesta datos en mini lotes y realiza transformaciones RDD (Resilient Distributed Datasets) en esos mini lotes de datos.

MLlib (biblioteca de aprendizaje automático)

MLlib es un marco de aprendizaje automático distribuido por encima de Spark debido a la arquitectura Spark basada en memoria distribuida. Según los puntos de referencia, los desarrolladores de MLlib lo hacen frente a las implementaciones de los mínimos cuadrados alternos (ALS). Spark MLlib es nueve veces más rápido que la versión basada en disco de Hadoop deApache Mahout (antes de que Mahout obtuviera una interfaz Spark).

GraphX

GraphX ​​es un marco de procesamiento de gráficos distribuido sobre Spark. Proporciona una API para expresar el cálculo de gráficos que puede modelar los gráficos definidos por el usuario mediante la API de abstracción de Pregel. También proporciona un tiempo de ejecución optimizado para esta abstracción.

Conjuntos de datos distribuidos resistentes

Los conjuntos de datos distribuidos resistentes (RDD) son una estructura de datos fundamental de Spark. Es una colección distribuida inmutable de objetos. Cada conjunto de datos en RDD se divide en particiones lógicas, que se pueden calcular en diferentes nodos del clúster. Los RDD pueden contener cualquier tipo de objetos Python, Java o Scala, incluidas las clases definidas por el usuario.

Formalmente, un RDD es una colección de registros particionada de solo lectura. Los RDD se pueden crear mediante operaciones deterministas en datos en almacenamiento estable u otros RDD. RDD es una colección de elementos tolerantes a fallas que se pueden operar en paralelo.

Hay dos formas de crear RDD: parallelizing una colección existente en su programa de conductor, o referencing a dataset en un sistema de almacenamiento externo, como un sistema de archivos compartidos, HDFS, HBase o cualquier fuente de datos que ofrezca un formato de entrada Hadoop.

Spark hace uso del concepto de RDD para lograr operaciones MapReduce más rápidas y eficientes. Primero, analicemos cómo se llevan a cabo las operaciones de MapReduce y por qué no son tan eficientes.

El intercambio de datos es lento en MapReduce

MapReduce se adopta ampliamente para procesar y generar grandes conjuntos de datos con un algoritmo distribuido paralelo en un clúster. Permite a los usuarios escribir cálculos en paralelo, utilizando un conjunto de operadores de alto nivel, sin tener que preocuparse por la distribución del trabajo y la tolerancia a fallas.

Desafortunadamente, en la mayoría de los marcos actuales, la única forma de reutilizar datos entre cálculos (por ejemplo, entre dos trabajos de MapReduce) es escribirlos en un sistema de almacenamiento estable externo (por ejemplo, HDFS). Aunque este marco proporciona numerosas abstracciones para acceder a los recursos computacionales de un clúster, los usuarios aún quieren más.

Ambos Iterative y Interactivelas aplicaciones requieren un intercambio de datos más rápido en trabajos paralelos. El intercambio de datos es lento en MapReduce debido areplication, serializationy disk IO. En cuanto al sistema de almacenamiento, la mayoría de las aplicaciones de Hadoop pasan más del 90% del tiempo realizando operaciones de lectura y escritura HDFS.

Operaciones iterativas en MapReduce

Reutilice los resultados intermedios en varios cálculos en aplicaciones de varias etapas. La siguiente ilustración explica cómo funciona el marco actual, mientras se realizan las operaciones iterativas en MapReduce. Esto incurre en gastos generales sustanciales debido a la replicación de datos, E / S de disco y serialización, lo que hace que el sistema sea lento.

Operaciones interactivas en MapReduce

El usuario ejecuta consultas ad-hoc en el mismo subconjunto de datos. Cada consulta hará la E / S del disco en el almacenamiento estable, que puede dominar el tiempo de ejecución de la aplicación.

La siguiente ilustración explica cómo funciona el marco actual mientras se realizan las consultas interactivas en MapReduce.

Uso compartido de datos con Spark RDD

El intercambio de datos es lento en MapReduce debido a replication, serializationy disk IO. La mayoría de las aplicaciones de Hadoop pasan más del 90% del tiempo realizando operaciones de lectura y escritura HDFS.

Al reconocer este problema, los investigadores desarrollaron un marco especializado llamado Apache Spark. La idea clave de la chispa esResiliente Datribuido Datasets (RDD); admite el cálculo de procesamiento en memoria. Esto significa que almacena el estado de la memoria como un objeto en los trabajos y el objeto se puede compartir entre esos trabajos. El intercambio de datos en la memoria es de 10 a 100 veces más rápido que la red y el disco.

Intentemos ahora descubrir cómo se llevan a cabo las operaciones iterativas e interactivas en Spark RDD.

Operaciones iterativas en Spark RDD

La ilustración que se muestra a continuación muestra las operaciones iterativas en Spark RDD. Almacenará los resultados intermedios en una memoria distribuida en lugar de un almacenamiento estable (disco) y hará que el sistema sea más rápido.

Note - Si la memoria distribuida (RAM) es suficiente para almacenar resultados intermedios (estado del trabajo), entonces almacenará esos resultados en el disco

Operaciones interactivas en Spark RDD

Esta ilustración muestra operaciones interactivas en Spark RDD. Si se ejecutan diferentes consultas en el mismo conjunto de datos repetidamente, estos datos en particular se pueden mantener en la memoria para mejorar los tiempos de ejecución.

De forma predeterminada, cada RDD transformado se puede volver a calcular cada vez que ejecuta una acción en él. Sin embargo, también puedepersistun RDD en la memoria, en cuyo caso Spark mantendrá los elementos en el clúster para un acceso mucho más rápido, la próxima vez que lo consulte. También es compatible con RDD persistentes en disco o replicados en varios nodos.

Spark es el subproyecto de Hadoop. Por lo tanto, es mejor instalar Spark en un sistema basado en Linux. Los siguientes pasos muestran cómo instalar Apache Spark.

Paso 1: verificar la instalación de Java

La instalación de Java es una de las cosas obligatorias al instalar Spark. Pruebe el siguiente comando para verificar la versión de JAVA.

$java -version

Si Java ya está instalado en su sistema, verá la siguiente respuesta:

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)

En caso de que no tenga Java instalado en su sistema, instale Java antes de continuar con el siguiente paso.

Paso 2: Verificación de la instalación de Scala

Debe utilizar el lenguaje Scala para implementar Spark. Entonces, verifiquemos la instalación de Scala usando el siguiente comando.

$scala -version

Si Scala ya está instalado en su sistema, verá la siguiente respuesta:

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

En caso de que no tenga Scala instalado en su sistema, continúe con el siguiente paso para la instalación de Scala.

Paso 3: descarga de Scala

Descargue la última versión de Scala visitando el siguiente enlace Descargar Scala . Para este tutorial, usamos la versión scala-2.11.6. Después de la descarga, encontrará el archivo tar de Scala en la carpeta de descarga.

Paso 4: Instalación de Scala

Siga los pasos que se indican a continuación para instalar Scala.

Extraiga el archivo tar de Scala

Escriba el siguiente comando para extraer el archivo tar de Scala.

$ tar xvf scala-2.11.6.tgz

Mover archivos de software Scala

Utilice los siguientes comandos para mover los archivos del software Scala al directorio respectivo (/usr/local/scala).

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

Establecer PATH para Scala

Utilice el siguiente comando para configurar PATH para Scala.

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

Verificación de la instalación de Scala

Después de la instalación, es mejor verificarlo. Utilice el siguiente comando para verificar la instalación de Scala.

$scala -version

Si Scala ya está instalado en su sistema, verá la siguiente respuesta:

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

Paso 5: descarga de Apache Spark

Descargue la última versión de Spark visitando el siguiente enlace Descargar Spark . Para este tutorial, estamos usandospark-1.3.1-bin-hadoop2.6versión. Después de descargarlo, encontrará el archivo tar de Spark en la carpeta de descarga.

Paso 6: instalación de Spark

Siga los pasos que se indican a continuación para instalar Spark.

Extracción de alquitrán de chispa

El siguiente comando para extraer el archivo tar de Spark.

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

Mover archivos de software Spark

Los siguientes comandos para mover los archivos de software Spark al directorio respectivo (/usr/local/spark).

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

Configurar el entorno para Spark

Agregue la siguiente línea a ~/.bashrcarchivo. Significa agregar la ubicación, donde se encuentra el archivo de software Spark a la variable PATH.

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

Utilice el siguiente comando para obtener el archivo ~ / .bashrc.

$ source ~/.bashrc

Paso 7: Verificación de la instalación de Spark

Escriba el siguiente comando para abrir Spark shell.

$spark-shell

Si Spark se instala correctamente, encontrará el siguiente resultado.

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 presenta un módulo de programación para el procesamiento de datos estructurados llamado Spark SQL. Proporciona una abstracción de programación llamada DataFrame y puede actuar como motor de consultas SQL distribuido.

Características de Spark SQL

Las siguientes son las características de Spark SQL:

  • Integrated- Mezcle sin problemas consultas SQL con programas Spark. Spark SQL le permite consultar datos estructurados como un conjunto de datos distribuidos (RDD) en Spark, con API integradas en Python, Scala y Java. Esta estrecha integración facilita la ejecución de consultas SQL junto con complejos algoritmos analíticos.

  • Unified Data Access- Cargue y consulte datos de una variedad de fuentes. Los Schema-RDD proporcionan una interfaz única para trabajar de manera eficiente con datos estructurados, incluidas tablas Apache Hive, archivos parquet y archivos JSON.

  • Hive Compatibility- Ejecute consultas de Hive sin modificar en almacenes existentes. Spark SQL reutiliza la interfaz de Hive y MetaStore, lo que le brinda compatibilidad total con los datos, consultas y UDF de Hive existentes. Simplemente instálelo junto con Hive.

  • Standard Connectivity- Conéctese a través de JDBC u ODBC. Spark SQL incluye un modo de servidor con conectividad JDBC y ODBC estándar de la industria.

  • Scalability- Utilice el mismo motor para consultas interactivas y largas. Spark SQL aprovecha el modelo RDD para admitir la tolerancia a fallas a mitad de la consulta, lo que le permite escalar también a trabajos grandes. No se preocupe por utilizar un motor diferente para los datos históricos.

Arquitectura Spark SQL

La siguiente ilustración explica la arquitectura de Spark SQL:

Esta arquitectura contiene tres capas, a saber, API de lenguaje, RDD de esquema y Fuentes de datos.

  • Language API- Spark es compatible con diferentes lenguajes y Spark SQL. También es compatible con estos lenguajes: API (python, scala, java, HiveQL).

  • Schema RDD- Spark Core está diseñado con una estructura de datos especial llamada RDD. Generalmente, Spark SQL funciona en esquemas, tablas y registros. Por lo tanto, podemos usar Schema RDD como tabla temporal. Podemos llamar a este esquema RDD como marco de datos.

  • Data Sources- Por lo general, la fuente de datos para Spark-core es un archivo de texto, un archivo Avro, etc. Sin embargo, las fuentes de datos para Spark SQL son diferentes. Esos son el archivo Parquet, el documento JSON, las tablas HIVE y la base de datos Cassandra.

Discutiremos más sobre estos en los capítulos siguientes.

Un DataFrame es una colección distribuida de datos, que se organiza en columnas con nombre. Conceptualmente, equivale a tablas relacionales con buenas técnicas de optimización.

Un DataFrame se puede construir a partir de una matriz de diferentes fuentes, como tablas de Hive, archivos de datos estructurados, bases de datos externas o RDD existentes. Esta API se diseñó para aplicaciones modernas de Big Data y ciencia de datos inspirándose enDataFrame in R Programming y Pandas in Python.

Características de DataFrame

Aquí hay un conjunto de algunas características de DataFrame:

  • Capacidad para procesar los datos en el tamaño de Kilobytes a Petabytes en un clúster de un solo nodo a un clúster grande.

  • Admite diferentes formatos de datos (Avro, csv, búsqueda elástica y Cassandra) y sistemas de almacenamiento (HDFS, tablas HIVE, mysql, etc.).

  • Optimización de vanguardia y generación de código a través del optimizador Spark SQL Catalyst (marco de transformación de árboles).

  • Se puede integrar fácilmente con todas las herramientas y marcos de Big Data a través de Spark-Core.

  • Proporciona API para programación en Python, Java, Scala y R.

SQLContext

SQLContext es una clase y se usa para inicializar las funcionalidades de Spark SQL. El objeto de clase SparkContext (sc) es necesario para inicializar el objeto de clase SQLContext.

El siguiente comando se usa para inicializar SparkContext a través de spark-shell.

$ spark-shell

De forma predeterminada, el objeto SparkContext se inicializa con el nombre sc cuando comienza la chispa.

Utilice el siguiente comando para crear SQLContext.

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

Ejemplo

Consideremos un ejemplo de registros de empleados en un archivo JSON llamado employee.json. Utilice los siguientes comandos para crear un DataFrame (df) y leer un documento JSON llamadoemployee.json con el siguiente contenido.

employee.json - Coloque este archivo en el directorio donde se encuentra scala> se encuentra el puntero.

{
   {"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"}
}

Operaciones de DataFrame

DataFrame proporciona un lenguaje específico de dominio para la manipulación de datos estructurados. Aquí, incluimos algunos ejemplos básicos de procesamiento de datos estructurados usando DataFrames.

Siga los pasos que se indican a continuación para realizar operaciones de DataFrame:

Leer el documento JSON

Primero, tenemos que leer el documento JSON. Basado en esto, genere un DataFrame llamado (dfs).

Use el siguiente comando para leer el documento JSON llamado employee.json. Los datos se muestran como una tabla con los campos: id, nombre y edad.

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

Output - Los nombres de los campos se toman automáticamente de employee.json.

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

Mostrar los datos

Si desea ver los datos en el DataFrame, utilice el siguiente comando.

scala> dfs.show()

Output - Puede ver los datos de los empleados en formato tabular.

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

Utilice el método printSchema

Si desea ver la estructura (esquema) del DataFrame, utilice el siguiente comando.

scala> dfs.printSchema()

Output

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

Usar método de selección

Utilice el siguiente comando para recuperar name-columna entre tres columnas del DataFrame.

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

Output - Puedes ver los valores de la name columna.

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

Usar filtro de edad

Utilice el siguiente comando para encontrar los empleados cuya edad sea superior a 23 (edad> 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  |
+----+------+--------+

Usar groupBy Method

Utilice el siguiente comando para contar el número de empleados que tienen la misma edad.

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

Output - dos empleados tienen 23 años.

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

Ejecución de consultas SQL mediante programación

Un SQLContext permite que las aplicaciones ejecuten consultas SQL mediante programación mientras ejecutan funciones SQL y devuelve el resultado como un DataFrame.

Generalmente, en segundo plano, SparkSQL admite dos métodos diferentes para convertir RDD existentes en DataFrames:

No Señor Métodos y descripción
1 Inferir el esquema usando la reflexión

Este método utiliza la reflexión para generar el esquema de un RDD que contiene tipos específicos de objetos.
2 Especificar el esquema mediante programación

El segundo método para crear DataFrame es a través de una interfaz programática que le permite construir un esquema y luego aplicarlo a un RDD existente.

Una interfaz DataFrame permite que diferentes fuentes de datos funcionen en Spark SQL. Es una mesa temporal y se puede operar como un RDD normal. El registro de un DataFrame como una tabla le permite ejecutar consultas SQL sobre sus datos.

En este capítulo, describiremos los métodos generales para cargar y guardar datos usando diferentes fuentes de datos de Spark. A continuación, discutiremos en detalle las opciones específicas que están disponibles para las fuentes de datos integradas.

Hay diferentes tipos de fuentes de datos disponibles en SparkSQL, algunas de las cuales se enumeran a continuación:

No Señor Fuentes de datos
1 Conjuntos de datos JSON

Spark SQL puede capturar automáticamente el esquema de un conjunto de datos JSON y cargarlo como un DataFrame.
2 Mesas Colmena

Hive viene incluido con la biblioteca Spark como HiveContext, que hereda de SQLContext.
3 Archivos de parquet

Parquet es un formato columnar, compatible con muchos sistemas de procesamiento de datos.