Cursos Empresas Blog Live Conf Eventos Precios

0:0012:17

Recursos

Resumen

¿Cómo trabajar con archivos y esquemas en Spark SQL?

Para triunfar en el manejo de datos, es esencial comprender cómo leer y manipular información adecuadamente. En el contexto de Spark SQL, aprender a importar y definir esquemas es el primer paso crucial. Comencemos cargando un archivo persona.data, al cual aplicamos un esquema de metadatos. Esto nos permite especificar claramente los tipos de datos para cada campo, como ID, nombre o correo electrónico.

from pyspark.sql.types import StructType, StructField, StringType

schema = StructType([
    StructField("ID", StringType(), True),
    StructField("Nombre", StringType(), True),
    StructField("Teléfono", StringType(), True),
    # Añade los demás campos aquí
])

dfPersona = spark.read.format("csv") \
    .option("header", "true") \
    .schema(schema) \
    .load("/FileStore-CursoDatabricks/persona.data")
dfPersona.show(10)

Este enfoque asegura que puedas manejar los datos adecuadamente, reduciendo riesgos de errores al definir claramente los tipos.

¿Cómo crear y manejar una vista temporal?

Para mejorar el manejo de consultas SQL en Spark, es esencial crear una vista temporal de los datos. Esto nos permite interactuar con los datos como si fueran una tabla en SQL. Crear una vista es sencillo:

dfPersona.createOrReplaceTempView("dfPersonaView")

Una vez creada, puedes ejecutar consultas SQL directamente sobre esta vista:

result = spark.sql("SELECT * FROM dfPersonaView WHERE edad > 30")
result.show(10)

Esto simplifica la aplicación de filtros y condiciones, maximizando la eficiencia en tu análisis de datos.

¿Cómo optimizar consultas largas en Spark SQL?

Las consultas extensas pueden complicarse y volverse difíciles de manejar. Aquí destacamos el uso de triple comillas para simplificar y estructurar consultas largas, haciéndolas más legibles:

query = """
SELECT id, nombre, correo
FROM dfPersonaView
WHERE edad > 30 AND salario > 5000
"""
dfResult = spark.sql(query)
dfResult.show(10)

Este método mejora la claridad del código, facilitando el entendimiento y mantenimiento de tus consultas.

¿Qué ventajas tiene la parametrización en consultas SQL?

Incorporar parámetros en tus consultas añade dinamismo y flexibilidad, permitiendo cambios sin modificar la lógica SQL:

paramEdad = 30
paramSalario = 5000

query = f"""
SELECT id, nombre, correo
FROM dfPersonaView
WHERE edad > {paramEdad} AND salario > {paramSalario}
"""
dfResult = spark.sql(query)
dfResult.show(10)

La parametrización permite una gestión más eficiente y adaptable de tus consultas, agilizando los procesos de modificación y mantenimiento.

¿Cómo practicar y aplicar estos conceptos?

Nada refuerza el aprendizaje más que la práctica continua. Aquí te presentamos tres ejercicios que te permitirán aplicar lo aprendido:

Seleccionar nombre y edad de personas mayores de 28 años.
Calcular la edad promedio en el DataFrame.
Extraer nombre, edad y dirección de personas con direcciones registradas.

Desafíate a resolver estos ejercicios utilizando las técnicas presentadas y fortalece tus habilidades en Spark SQL. ¡Sigue explorando, aprendiendo y dominando este vasto mundo de la gestión de datos!

Consultas y Filtrado Avanzado con Spark SQL

Curso de Databricks: Arquitectura Delta Lake

Clase 17 de 25

Clase 17 / 25

Paula Andrea Chamorro Rodríguez

Hace 3 años

Hola

En Python utilizo df.info() para poder ver rápidamente si tengo nulos en cada una de las columnas del df. Hay algunas función en Pyspark que me permita hacer esta revisión?

Layla Scheli

Profesor

Hace 3 años

Hola Paula, sí, en PySpark puedes utilizar la función df.printSchema() para obtener información sobre el esquema del DataFrame, incluyendo los tipos de datos y si las columnas permiten valores nulos o no. Además, también puedes usar df.describe() para obtener estadísticas descriptivas de las columnas numéricas, lo que podría ayudarte a identificar posibles valores nulos en esas columnas si ves que el recuento de valores no es consistente.

Sin embargo, para verificar específicamente la presencia de valores nulos en cada columna, puedes usar el método isNull() junto con la función df.select() para contar los valores nulos en cada columna.

from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum as spark_sum

Inicializar SparkSession

spark = SparkSession.builder
.appName("NullCheck")
.getOrCreate()

Crear un DataFrame de ejemplo

data = [("John", 30, None), ("Alice", None, 70), ("Bob", 25, 80)]

columns = ["Name", "Age", "Score"]

df = spark.createDataFrame(data, columns)

Contar los valores nulos en cada columna

null_counts = df.select(*(spark_sum(col(c).isNull().cast("int")).alias(c) for c in df.columns))

Mostrar el resultado

null_counts.show()

Ahi va un ejemplo :) Un saludo!

Comentarios (16)16 comentarios

Christian Julian Acosta Santamaria

Gracias Christian :) lo validamos. Un saludo!

Enrique Alejandro Terrazas Huamaní

Hace 2 años

Gracias, me fue útil tu explicación

Yhomira Alexandra Yupayccana Lopa

Mario Alexander Vargas Celis

¡Entendido! Si estás trabajando en un laboratorio de PySpark SQL, seguramente estás explorando cómo usar las capacidades SQL en PySpark para consultas, análisis y manipulación de datos.

Parte 2: Ejemplo estructurado para PySpark SQL

A continuación, te guiaré a través de una estructura típica para trabajar con PySpark SQL en un laboratorio:

1. Crear o cargar un DataFrame

Cargamos datos desde un archivo CSV o creamos un DataFrame manualmente.

from pyspark.sql import SparkSession from pyspark.sql.functions import col

# Crear una SparkSession spark = SparkSession.builder.appName("PySpark SQL Lab").getOrCreate()

# Crear un DataFrame manualmente data = [ (1, "Alice", 5000), (2, "Bob", 4000), (3, "Charlie", 7000), (4, "David", 10000), ] columns = ["id", "name", "salary"]

# Mostrar los datos iniciales df.show()

2. Registrar una vista temporal

Registrar una tabla temporal para ejecutar consultas SQL.

# Registrar el DataFrame como una tabla temporal df.createOrReplaceTempView("employees")

3. Ejecutar consultas SQL

Usar Spark SQL para ejecutar consultas sobre los datos.

# Ejemplo 1: Seleccionar todos los registros result = spark.sql("SELECT * FROM employees") result.show()

# Ejemplo 2: Filtrar empleados con salario mayor a 5000 result = spark.sql("SELECT * FROM employees WHERE salary > 5000") result.show()

# Ejemplo 3: Agregar una columna calculada (bonus) result = spark.sql("SELECT *, salary * 0.1 AS bonus FROM employees") result.show()

# Ejemplo 4: Calcular la suma total de salarios result = spark.sql("SELECT SUM(salary) AS total_salary FROM employees") result.show()

4. Realizar transformaciones adicionales

Combinar operaciones SQL y funciones PySpark.

# Agregar una columna 'bonus' al DataFrame original y calcular el total df = df.withColumn("bonus", col("salary") * 0.1) df = df.withColumn("total", col("salary") + col("bonus"))

# Mostrar el DataFrame actualizado df.show()

5. Guardar los resultados

Guardar el resultado de las consultas o transformaciones en diferentes formatos.

# Guardar el DataFrame actualizado como un archivo CSV df.write.csv("output/employees_with_bonus", header=True)

John Arango

Ejercicio 2

from pyspark.sql import functions as f
#Ejercicio 2 - Resolucion
(df.agg(f.avg("edad")).show())
# Completar

Ejercicio 3

#Ejercicio 3 - Resolucion
df.join(df_direccion, df.id == df_direccion.id).select("nombre", "edad", "direccion").show()
# Completar

Libardo Jose Castro Aparicio

David Cuellar

Cristian Camilo Cardenas Rodríguez

Nota: para el punto 2 se debe crear una nueva vista o sobreescribir la anterior ya que no se veria la agregación de la fila en el ejercicio 2

Angello Triviño Umaña

Resolución ejercicio 3: usando sintaxis SQL y df.join

Javier Ladino

Hace 9 meses

Juan Camilo Sánchez

Hace un año

There I go

Gerardo Mayel Fernández Alamilla

Tengo una duda, cómo funciona en memoria lo que acabamos de hacer veo que primero creamos un object de spray y luego en otro que emula una tabla SQL, esos dos objetos están en la RAM de la instancia? es común hacer esto en un conjunto de datos masivo?

Hola Gerardo, si para SPARKSQL lo necesitas de esa manera.

Aca tenes mas info :)

https://programmerclick.com/article/671786660/

Gabriela Andreina García Uzcategui

Mis resultados:

Jesús David Vega Porras

Aqui mis respuestas:#Ejercicio 1 - Resolucionparam_edad = 28spark.sql(f"""SELECT nombre, edad FROM personas WHERE edad > {param_edad} """).show()#Ejercicio 2 - Resolucion#Ejercicio 3 - Resolucion spark.sql("""SELECT p.nombre, p.edad, d.direccion FROM personas p JOIN direcciones d ON p.id = d.id""").show() df.agg({'edad': 'avg'}).show()

#Ejercicio 1 - Resolucion
param_edad = 28
spark.sql(f"""SELECT nombre, edad
            FROM personas
            WHERE
            edad > {param_edad}
          """).show()   
``````python
#Ejercicio 2 - Resolucion

df.agg({'edad': 'avg'}).show()
``````js
#Ejercicio 3 - Resolucion

spark.sql("""SELECT p.nombre, p.edad, d.direccion
            FROM personas p
            JOIN direcciones d ON p.id = d.id""").show()

Nicolás Mantilla

Hola les dejo la solución a los ejercicios

Punto 1

punto 2

punto 3

resultado

Fernando Ortiz

Les comparto mis resultados:

Es necesario remplazar la vistas

Saludos.

Gracias :)

Recursos

Resumen

Consultas y Filtrado Avanzado con Spark SQL

Progreso del curso

Fundamentos del Data Management con Databricks

Administración y Gestion de la Plataforma de Databricks

Apache Spark SQL y UDF

Implementacion de un Delta Lake en Databricks