Ventas por país: tres mil millones de filas
Cada lunes, una cadena de tiendas quiere una sola cifra: sus ventas por país. La tabla de pedidos tiene tres mil millones de filas, repartidas en veinte máquinas.
No quieres escribir veinte programas. Quieres describir el resultado una vez y dejar que el sistema reparta el trabajo.
Al final de esta lección
Nota
Puedo expresar un tratamiento de datos como una tubería (pipeline) de Spark.
Palabras clave
| Palabra | Qué significa | Ejemplo |
|---|---|---|
| DataFrame | una tabla de filas y columnas con nombre sobre la que trabaja Spark | |
| Transformación | un paso que fabrica una tabla nueva a partir de otra | filter |
| Acción | un paso que ejecuta el plan y da una respuesta | count |
| Perezoso | que espera a que pidas una respuesta antes de trabajar (evaluación perezosa, lazy en inglés) | |
| Partición | un trozo de una tabla, en manos de un solo trabajador | |
| Trabajo | lo que hace Spark para responder a una acción (job en inglés) |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Un DataFrame es una tabla que se consulta
MapReduce te obliga a escribir los pasos map y reduce. Spark te deja describir el resultado sobre un DataFrame: una tabla con columnas que tienen nombre.
Consultas un DataFrame con código, como filter y groupBy, o con SQL. Los dos dan el mismo plan. El laboratorio ha cargado un archivo de 24 pedidos con el nombre orders.
Cómo responde el laboratorio
El terminal del laboratorio no tiene Spark. Envía tus preguntas con curl a un pequeño servicio, que siempre responde con una tabla de texto.
Los nombres de las columnas vienen del archivo, así que están en inglés: country, product, quantity. Este servicio muestra incluso el resultado de una transformación sola, y eso obliga a Spark a trabajar.
Mirar cuatro filas
curl -s -d 'SELECT * FROM orders LIMIT 4' http://spark:8000/sqlSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Contar los países distintos
curl -s -d 'orders.select("country").distinct().count()' http://spark:8000/dfSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Las transformaciones esperan, las acciones lanzan
Una transformación como filter o groupBy fabrica un DataFrame nuevo, pero todavía no lee nada. Spark solo añade el paso a su plan. Por eso decimos que es perezoso.
Una acción como count o collect pide una respuesta. Entonces Spark ejecuta todo el plan: eso es un trabajo.
Ejemplo: pasos y su tipo
| Paso | Tipo | ¿Lee los datos? |
|---|---|---|
| filter(orders.quantity > 3) | Transformación | No, se añade al plan |
| select("country") | Transformación | No |
| groupBy("country").sum("quantity") | Transformación | No |
| count() | Acción | Sí, ejecuta el plan |
| collect() | Acción | Sí, y trae todas las filas |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Un plan que se ejecuta cuando se pide
# Un plan es una lista de pasos. Añadir un paso no lee ningún dato.
rows = [["Francia", 2], ["Japón", 1], ["Francia", 5]]
plan = []
plan.append(["keep_more_than", 1])
print("Plan:", plan)
def run_plan(rows, plan):
# Una acción ejecuta todo el plan sobre las filas.
for step in plan:
if step[0] == "keep_more_than":
kept = []
for row in rows:
if row[1] > step[1]:
kept.append(row)
rows = kept
return rows
print("Resultado:", run_plan(rows, plan))
# Cambia el 1 del plan por un 2 y vuelve a ejecutar.
Ejecutar un plan: los pedidos grandes
curl -s -d 'orders.filter(orders.quantity >= 5).count()' http://spark:8000/dfSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Un error muy común
Advertencia
Mucha gente cree que una línea filter lee los datos enseguida. No es así: solo añade un paso al plan. Spark sí comprueba al momento que las columnas existen. Un valor erróneo dentro de los datos solo aparece cuando una acción ejecuta el plan.
Las particiones reparten el trabajo
Spark parte un DataFrame en particiones. Cada trabajador suma la suya. Después Spark suma las sumas parciales.
Es otra vez la idea de map y reduce, sin que la escribas tú. La interfaz Spark UI muestra cada trabajo y su plan: ves lo que se ha ejecutado.
Sumas parciales y un total
# Tres particiones de la columna de cantidades, cada una en otro trabajador.
partitions = [[3, 1, 2], [4, 2], [6, 1]]
partial_sums = []
for part in partitions:
total = 0
for quantity in part:
total = total + quantity
partial_sums.append(total)
print("Sumas parciales:", partial_sums)
grand_total = 0
for value in partial_sums:
grand_total = grand_total + value
print("Total:", grand_total)
# Mueve el 6 a la primera partición y vuelve a ejecutar.
Preguntar cuántas particiones tiene orders
curl -s -d 'orders.rdd.getNumPartitions()' http://spark:8000/dfSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Partir orders en tres y contar cada trozo
curl -s -d 'orders.repartition(3).rdd.glom().map(len).collect()' http://spark:8000/dfSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Pedidos recibidos en cada país
curl -s -d 'SELECT country, count(*) AS orders_count FROM orders GROUP BY country ORDER BY orders_count DESC, country' http://spark:8000/sqlSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Spark UI muestra cada trabajo
Abre Spark UI desde la barra del laboratorio. La pestaña Jobs lista el trabajo hecho para cada acción. La pestaña SQL / DataFrame lista cada consulta con su plan.
Repite un comando de arriba y mira cómo aparece una fila nueva. Así averigua un equipo de datos por qué una tubería va lenta.
Para ir más lejos: la media por país
Escribe tu propia consulta SQL para el servicio /sql. Para cada país, debe dar el número de pedidos y la cantidad media por pedido, de la media mayor a la menor.
Antes de ejecutarla, apuesta qué país sale primero. Después abre Spark UI y busca el trabajo que ha creado tu consulta.