¿Hace falta un clúster para esto?
Spark, SQL y Python simple son tres maneras de hacer el mismo trabajo con una tabla de datos. La mejor depende de lo grandes que sean los datos.
Un equipo quiere la facturación de cada país. Una persona piensa en Spark. Otra dice que un bucle de Python de diez líneas bastaría. Hoy haces el mismo trabajo de tres maneras y decides con una razón.
Al final de esta lección
Nota
Puedo hacer un mismo trabajo con un DataFrame de Spark, con SQL y con Python simple, y decir cuándo conviene cada uno.
Palabras clave
| Palabra | Qué significa | Ejemplo |
|---|---|---|
| DataFrame | una tabla de filas y columnas con nombre sobre la que trabaja Spark | |
| Spark SQL | hacer una pregunta a un DataFrame en SQL | |
| Redistribución | filas que pasan de un trabajador a otro para que las claves iguales se junten | Exchange |
| Plan físico | la lista de pasos que Spark va a ejecutar de verdad | |
| En memoria | tan pequeño que cabe en la memoria de una máquina |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Un trabajo, tres maneras
El trabajo es la facturación de cada país: cantidad por precio, sumado para cada país. El laboratorio tiene 24 pedidos cargados como orders.
Primero como expresión de DataFrame de Spark, luego como SQL de Spark, luego como bucle de Python simple sobre el mismo archivo. Si las tres no coinciden, una se equivoca.
La manera DataFrame
curl -s -d 'orders.groupBy("country").agg(F.round(F.sum(orders.quantity * orders.unit_price), 2).alias("revenue")).orderBy("country")' 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.
La manera SQL
curl -s -d 'SELECT country, round(sum(quantity * unit_price), 2) AS revenue FROM orders GROUP BY country ORDER BY 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.
La manera Python simple
curl -s -d 'import csv, collections; c = collections.Counter(); [c.update({r["country"]: float(r["quantity"]) * float(r["unit_price"])}) for r in csv.DictReader(open("/tmp/lab/orders.csv"))]; sorted(c.items())' http://spark:8000/pySe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
El mismo trabajo como bucle
# El mismo trabajo como bucle de Python simple: la facturación de cada país.
orders = [["UK", 2, 25.0], ["Japan", 1, 150.0], ["UK", 1, 150.0], ["Japan", 4, 12.5]]
revenue = {}
for order in orders:
country = order[0]
money = order[1] * order[2]
if country in revenue:
revenue[country] = revenue[country] + money
else:
revenue[country] = money
print(revenue)
# Añade ["France", 3, 12.5] a orders y vuelve a ejecutar.
Un DataFrame no es una lista
Un DataFrame parece una tabla que se recorre, pero es un plan. No guarda las filas en tu programa, así que muchos trucos de listas fallan.
Pídele su tamaño con count(), que es una acción. Pídele la longitud con len() y Python se niega.
Pedir la longitud como a una lista
curl -s -d 'len(orders)' 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.
Pedir el tamaño como a Spark
curl -s -d 'orders.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.
El plan muestra por dónde pasan los datos
Spark puede imprimir el plan físico que va a ejecutar. La palabra que hay que buscar es Exchange. Es una redistribución: las filas viajan entre trabajadores para que las claves iguales se junten.
Una redistribución es la parte cara de un trabajo grande. Leer un plan dice dónde está.
Un plan físico, como lo imprime Spark
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#19], functions=[count(1)])
+- Exchange hashpartitioning(country#19, 4), ENSURE_REQUIREMENTS, [plan_id=244]
+- HashAggregate(keys=[country#19], functions=[partial_count(1)])
+- InMemoryTableScan [country#19]
Listar los operadores principales del plan
curl -s -d 'orders.groupBy("country").count()._jdf.queryExecution().simpleString()' http://spark:8000/df | grep -oE "Exchange|HashAggregate|FileScan"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Un error frecuente
Advertencia
Mucha gente cree que Spark es siempre la mejor herramienta porque es la más grande. Con 24 filas solo añade un arranque y un plan. Spark compensa cuando los datos ya no caben en una máquina o el trabajo hay que repartirlo.
Ejemplo resuelto: elegir una herramienta
| Los datos | Una elección razonable | Por qué |
|---|---|---|
| 24 filas en un archivo | Un bucle de Python simple | Nada que repartir |
| 2 millones de filas que caben en memoria | pandas, o SQL en una base de datos | Una máquina lo hace en segundos |
| Miles de millones de filas en muchas máquinas | Spark | El trabajo hay que repartirlo |
| Un trabajo que se ejecuta cada noche | El que esté escrito con claridad y comprobado | La misma respuesta cada vez |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Para ir más lejos: otra pregunta
Calcula las unidades vendidas de cada producto con la manera DataFrame, la manera SQL y la manera Python simple. ¿Coinciden?
Pista: agrupa por product y suma quantity. Luego imprime el plan de tu línea de DataFrame y busca su Exchange.