Uso de SQLContext para Registro y Consultas con DataFrames en Spark

Clase 19 de 25 • Curso de Fundamentos de Spark para Big Data

Resumen

¿Cómo registrar data frames como SQL?

Integrar Spark con SQL puede ser un poderoso recurso para los amantes de SQL que desean aprovechar sus habilidades en el análisis y gestión de datos grandes. Aquí, aprenderemos a registrar data frames como tablas temporales y utilizar SQL de una manera práctica y eficiente.

Registro de data frames como tablas temporales

Para emplear data frames como tablas en SQL, primero debemos registrarlos como tablas temporales. En nuestro ejemplo, registraremos tres data frames: resultado, deportista olímpico y países. Usamos el método register con un alias que simplifica su uso posterior.

# Registro de data frames como tablas temporales
resultado.registerTempTable("resultado")
deportista_olímpico.registerTempTable("deportista")
países.registerTempTable("paises")

Uso del SQLContext

El objeto SQLContext en Spark nos permite ejecutar comandos SQL directamente. Puedes especificar las sentencias SQL respetando el uso de palabras reservadas en mayúsculas por buenas prácticas. Por ejemplo:

# Ejecución de una consulta SQL simple
sqlContext.sql("SELECT medalla, equipo, sigla FROM resultado").show(5)

Comparación con el enfoque nativo de Spark

Aunque SQL es poderoso y familiar para muchos, el enfoque nativo de Spark ofrece ventajas significativas en rendimiento y eficiencia. A continuación, se realiza un ejemplo de consulta más complejo usando SQL, realizando JOINs entre tablas y aplicando restricciones.

SELECT medalla, equipo, sigla 
FROM resultado AS r
JOIN deportista AS d ON r.deportistaID = d.deportistaID
JOIN paises AS p ON d.equipoID = p.ID
WHERE medalla IS NOT NULL
ORDER BY sigla DESC

¿Cuándo usar SQL y cuándo funciones nativas de Spark?

SQL: Ideal para ejecuciones cortas debido a su familiaridad y simplicidad en el código.
Funciones nativas de Spark: Se recomienda al trabajar con grandes volúmenes de datos, ya que maximizan el rendimiento utilizando optimizaciones de ejecución específicas de Spark.

Consejos prácticos para usuarios avanzados

Aprovecha el optimizador de datos de Spark: Este controla el desempeño y corrige microerrores en ejecución.
Evalúa el contexto: Elige SQL para tareas más pequeñas o cuando la familiaridad es clave, pero cambia a las funciones de Spark para un mayor rendimiento con big data.

En resumen, elegir entre SQL y las funciones nativas de Spark depende del contexto y el tamaño de las operaciones. Es fundamental identificar cuál herramienta se alinea mejor con las necesidades de procesamiento sin comprometer la eficiencia. ¡Sigue explorando Spark y SQL para mejorar tus capacidades analíticas y de procesamiento de datos!

Juan Diego Gallego Villada

student•

Saludos! El método .registerTempTable("[alias]") está deprecado para las nuevas versiones. En ese caso se usa el método .createOrReplaceTempView("[alias]").

Jose Daniel Gallegos Padilla

student•

Vi que esta mostrando duplicados por lo que agregue la palabra DISTINCT despues del SELECT para eliminar los duplicados.

sqlContext.sql("""SELECT DISTINCT medalla, equipo, sigla
                    FROM resultado r
                    JOIN deportista d
                      ON r.deportista_id = d.deportista_id
                    JOIN paises p
                      ON p.equipo_id = d.equipo_id
                   WHERE medalla <> "NA"
                   ORDER BY sigla DESC
                    """).show()

Hubert Ronald Mendoza Canales

student•

Adicional a los tipos de medallas que ha ganado un país a lo largo de su historia, se está interesado en saber cuántas de éstas fueron, se puede emplear GROUP BY y la función COUNT

sqlContext.sql("""
    SELECT 
        medalla,
        equipo,
        sigla,
        COUNT(medalla) AS total_medalla
    FROM resultado r
    JOIN deportista d
    ON r.deportista_id = d.deportista_id
    JOIN paises p
    ON p.equipo_id = d.equipo_id
    WHERE medalla <> 'NA'
    GROUP BY medalla, equipo, sigla
    ORDER BY sigla DESC
""").show(10)

+-------+--------------------+-----+-------------+
|medalla|              equipo|sigla|total_medalla|
+-------+--------------------+-----+-------------+
| Bronze|            Zimbabwe|  ZIM|            1|
|   Gold|            Zimbabwe|  ZIM|           17|
| Silver|            Zimbabwe|  ZIM|            4|
| Bronze|              Zambia|  ZAM|            1|
| Silver|              Zambia|  ZAM|            1|
| Silver|          Yugoslavia|  YUG|          180|
|   Gold|          Yugoslavia|  YUG|          136|
| Bronze|          Yugoslavia|  YUG|          101|
| Bronze|West Indies Feder...|  WIF|            1|
| Silver|             Vietnam|  VIE|            3|
+-------+--------------------+-----+-------------+
only showing top 10 rows

También es posible saber cuantos países han participado y que han ganado medalla, sin que se repitan usando el DISTINCT dentro del COUNT

sqlContext.sql("""
    SELECT 
        'Paises con Melladas' AS estatus,
        COUNT(DISTINCT sigla) AS total_medalla
    FROM resultado r
    JOIN deportista d
    ON r.deportista_id = d.deportista_id
    JOIN paises p
    ON p.equipo_id = d.equipo_id
    WHERE medalla <> 'NA'
    
    UNION ALL
    
    SELECT 
        'Paises Participantes' AS estatus,
        COUNT(DISTINCT sigla) AS total_medalla
    FROM resultado r
    JOIN deportista d
    ON r.deportista_id = d.deportista_id
    JOIN paises p
    ON p.equipo_id = d.equipo_id
""").show(10)

+--------------------+-------------+
|             estatus|total_medalla|
+--------------------+-------------+
| Paises con Melladas|          150|
|Paises Participantes|          228|
+--------------------+-------------+

KEVIN XAVIER FREGOSO ROMERO

student•

para los que quieren conectar spark con postgresql en ubuntu yo hize lo siguiente baje el driver de JDBC https://jdbc.postgresql.org/download.html descomprimi el archivo lo copie en descargas y lo pegue en la siguiente ruta /usr/share/java/ quedando usr/share/java/postgresql-42.3.1.jar, luego exporte la ruta al CLASSPATH con el siguiente comando (export CLASSPATH="$CLASSPATH:/usr/share/java/postgresql-42.3.1.jar"), y quedo listo para usar. despues en jupyter notebook escribi el siguiente codigo para conectar a la base

<code> 
from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.jars", "/usr/share/java/postgresql-42.3.1.jar") \
    .getOrCreate()

df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/nombredelabase") \
    .option("dbtable", "nombredelatabla") \
    .option("user", "postgres") \
    .option("password", "password") \
    .option("driver", "org.postgresql.Driver") \
    .load()

df.printSchema()

espero les sirva

Brayan Betancourt Villegas

student•

Pensé que cuando iba a correr el DF iba a tener una esquema mas simple, pero es todo lo contrario :

DF( Duración :1.3 min):

SQL( Duración : 7 s):

Entonces si usar la sintaxis de SQL tiene mas coste computacional , como puedo evidenciar esto ?

Elberth Ramón Cabrales Gutierrez

student•

Por que no puedo hacer joins? ya intente con un join sencillo pero me sale este error:

<Py4JJavaError                             Traceback (most recent call last)
<ipython-input-467-cc70e4482d2a> in <module>
----> 1 sqlContext.sql("""
      2                 SELECT nombre FROM resultado r
      3                 JOIN deportista d
      4                 ON r.deportista_id = d.deportista_id
      5 """).show()>

<Py4JJavaError: An error occurred while calling o5965.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 428.0 failed 1 times, most recent failure: Lost task 0.0 in stage 428.0 (TID 582) (192.168.1.72 executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):>

Carlos Eduardo Gomez García

teacher•

¡Hola! Me parece que tu consulta está mal, te faltan los "AS", así:

SELECT nombre FROM resultado AS r
JOIN deportista AS d
ON r.deportista_id = d.deportista_id

Fabio Gomez

student•

Hola, no logro entender el metodo registerTempTable. Solo crea una tabla para usar sql y se elmina cuando se cierra el contexto?

Alex Camacho

teacher•

Hola Fabio, En este link explican un poco al respecto de registerTempTable, espero te sea de utilidad.

Julián Esteban Oliveros Forero

student•

Para comparar

https://herovired.com/learning-hub/blogs/dataframe-vs-spark-sql-differences-and-comparison/#:~:text=It%20is%20easier%20to%20perform,the%20lack%20of%20distributed%20processing.

Carlos Javier Guevara Contreras

student•

interesante