Cuando el archivo ya no entra

En el capítulo 18 medimos dónde está el listón y llegamos a que casi nadie lo cruza. Este es la otra mitad: y si lo cruzas, qué.

Porque el capítulo anterior nombraba Spark y Hadoop y no te enseñaba a correr ninguno, que es como decirte que existe el mar 🌊

El problema, en una línea

Las herramientas que cargan el archivo entero en memoria antes de hacer nada tienen un techo: el de la máquina. Y ese techo llega antes de lo que parece, porque un archivo ocupa varias veces más en memoria que en disco.

La culpa la tienen sobre todo las columnas de texto. Míralo con SQL:

import os
import tempfile
import duckdb

con = duckdb.connect()
print(con.sql("""
SELECT count(*) AS filas,
       sum(length(ciudad) + length(segmento) + length(canal)
           + length(categoria)) AS letras_de_texto
FROM 'ventas-miss-yera.csv'
"""))

┌───────┬─────────────────┐
│ filas │ letras_de_texto │
│ int64 │     int128      │
├───────┼─────────────────┤
│  3037 │           92401 │
└───────┴─────────────────┘

92.401 letras en cuatro columnas. En el archivo cada una es un byte. Cargadas en memoria por una herramienta de análisis, cada texto pasa a ser un objeto con su propia cabecera, y ahí es donde se multiplica.

Ese multiplicador está medido en el capítulo de pandas de mi libro de Python. Acá lo que importa es la consecuencia: con un archivo de 5 GB en disco puedes necesitar una máquina de 30 para abrirlo, y con SQL no.

DuckDB: SQL contra el archivo, sin cargarlo

print(con.sql("""
SELECT canal, count(*) AS ventas, round(avg(compro), 3) AS conversion
FROM 'ventas-miss-yera.csv'
GROUP BY canal ORDER BY ventas DESC
"""))

┌─────────────┬────────┬────────────┐
│    canal    │ ventas │ conversion │
│   varchar   │ int64  │   double   │
├─────────────┼────────┼────────────┤
│ Web         │    801 │      0.583 │
│ Marketplace │    776 │      0.461 │
│ Tienda      │    735 │      0.619 │
│ WhatsApp    │    725 │      0.652 │
└─────────────┴────────┴────────────┘

Mira el FROM 'ventas-miss-yera.csv'. El archivo es la tabla. No hubo que crear una base, ni importar, ni definir columnas.

Y el SQL es el mismo que llevas quince capítulos escribiendo: el GROUP BY del capítulo 6 funciona igual sobre un archivo suelto.

DuckDB se instala con pip install duckdb y ya está. Sin servidor, sin Java, sin configuración. Es a las bases analíticas lo que SQLite es a las transaccionales.

Ahora uno grande de verdad

Voy a fabricar un archivo de casi un millón de filas repitiendo el mío:

carpeta = tempfile.mkdtemp()
grande = os.path.join(carpeta, 'grande.parquet')

con.sql("COPY (SELECT unidades, monto, canal, ciudad, i AS repeticion "
        "FROM 'ventas-miss-yera.csv', range(1, 300) t(i)) "
        "TO '" + grande + "' (FORMAT parquet)")

print('filas:', con.sql("SELECT count(*) FROM '" + grande + "'").fetchone()[0])
print('pesa :', round(os.path.getsize(grande) / 1024 / 1024, 2), 'MB')

filas: 908063
pesa : 0.51 MB

908.063 filas en medio mega. Ese es el segundo tema del capítulo.

Parquet, y por qué pesa cincuenta veces menos

csv_grande = os.path.join(carpeta, 'grande.csv')
con.sql("COPY (SELECT * FROM '" + grande + "') TO '" + csv_grande
        + "' (FORMAT csv, HEADER)")

print('parquet:', round(os.path.getsize(grande) / 1024 / 1024, 2), 'MB')
print('csv    :', round(os.path.getsize(csv_grande) / 1024 / 1024, 2), 'MB')
print('veces mas chico:', round(os.path.getsize(csv_grande)
                                / os.path.getsize(grande), 1))

parquet: 0.51 MB
csv    : 25.4 MB
veces mas chico: 49.7

Cincuenta veces. Los mismos datos.

La razón es que un CSV guarda por filas y Parquet guarda por columnas. Y una columna entera se parece mucho a sí misma: si la ciudad "Arequipa" se repite 164.450 veces, Parquet la escribe una vez y anota cuántas. Un CSV la escribe 164.450 veces.

Y trae una segunda ventaja que importa todavía más: si tu consulta pide dos columnas de catorce, Parquet lee solo esas dos del disco. Un CSV hay que recorrerlo entero para llegar a la columna del final.

Si te llevas una sola cosa práctica de este capítulo, que sea esta: cuando un CSV te empiece a molestar, conviértelo a Parquet antes de comprar más memoria 🧾

Consultarlo sin cargarlo

print(con.sql("SELECT ciudad, count(*) AS filas FROM '" + grande
              + "' GROUP BY ciudad ORDER BY 2 DESC LIMIT 5"))

┌──────────┬────────┐
│  ciudad  │ filas  │
│ varchar  │ int64  │
├──────────┼────────┤
│ Arequipa │ 164450 │
│ Piura    │ 154583 │
│ Chiclayo │ 153088 │
│ Cusco    │ 150696 │
│ Trujillo │ 143520 │
└──────────┴────────┘

Casi un millón de filas agrupadas, y en ningún momento el archivo entero estuvo en memoria. DuckDB leyó la columna que necesitaba, por trozos, y fue sumando.

Eso se llama procesamiento por streaming, y es la diferencia de fondo con pandas: pandas quiere el archivo antes de empezar, y esto empieza y va leyendo.

Entonces, ¿dónde entra Spark?

Acá es donde quiero ser clara, porque es lo que el capítulo anterior te debía.

DuckDB corre en una máquina. Aprovecha todos sus núcleos y lee del disco por trozos, y por eso aguanta archivos de decenas o cientos de gigas en una laptop decente.

Spark reparte el trabajo entre muchas máquinas. Parte el archivo, manda un pedazo a cada una, cada una calcula lo suyo y después se juntan los resultados. Esa es toda la idea, y viene de un artículo de Google de 2004 que se llamaba MapReduce, que es lo que Hadoop implementó primero.

Y la parte incómoda: Spark tiene un coste fijo. Repartir, mandar y volver a juntar tarda, y con archivos que caben en una máquina esa gestión pesa más que el cálculo. Es normal ver a Spark perder contra DuckDB en datos de unos pocos gigas.

El SQL, eso sí, es casi el mismo. Si sabes escribir esta consulta, sabes escribirla en Spark 🐣

El error que sale al primer intento

con.sql("SELECT segmento FROM '" + grande + "' LIMIT 1")

BinderException: Binder Error: Referenced column "segmento" not found in FROM clause!
Candidate bindings: "monto", "repeticion"

Cuando escribí el Parquet me llevé solo cinco columnas, así que el segmento no está ahí. Y fíjate en la segunda línea: DuckDB te sugiere las que más se parecen a lo que escribiste, que es de las cosas más cómodas que tiene.

Y esto es un aviso de algo más grande: al convertir a Parquet uno decide qué columnas viajan, y esa decisión se olvida. Tres meses después alguien pide un corte por segmento y hay que rehacer la conversión desde el archivo original, que ojalá alguien haya guardado.

Lo que te llevas

• pandas carga el archivo entero; un CSV ocupa varias veces más en memoria que en disco.
• DuckDB consulta el archivo directamente con SQL, sin importar nada.
• Parquet guarda por columnas y suele pesar decenas de veces menos.
• Parquet lee solo las columnas que pides.
• DuckDB usa una máquina; Spark reparte entre varias.
• Spark tiene un coste fijo de repartir, y con pocos gigas suele perder.
• El SQL es casi el mismo en los tres sitios.

Ejercicios

una = con.sql("SELECT count(DISTINCT ciudad) FROM '" + grande + "'").fetchone()
todas = con.sql("SELECT count(*) FROM (SELECT * FROM '" + grande + "')").fetchone()
print('ciudades distintas:', una[0])
print('filas totales     :', todas[0])

ciudades distintas: 9
filas totales     : 908063

print(con.sql("SELECT ciudad, round(avg(TRY_CAST(monto AS DOUBLE)), 2) AS ticket, "
              "count(*) AS n FROM '" + grande + "' WHERE canal = 'WhatsApp' "
              "GROUP BY ciudad HAVING count(*) > 20000 ORDER BY ticket DESC"))

┌──────────┬────────┬───────┐
│  ciudad  │ ticket │   n   │
│ varchar  │ double │ int64 │
├──────────┼────────┼───────┤
│ Chiclayo │ 892.98 │ 34385 │
│ Piura    │ 805.25 │ 38571 │
│ Arequipa │ 761.19 │ 39767 │
│ Trujillo │ 726.09 │ 33189 │
│ Cusco    │ 684.33 │ 33488 │
└──────────┴────────┴───────┘

print(con.sql("""
SELECT
  sum(length(ciudad) + length(segmento) + length(canal)
      + length(categoria)) AS bytes_de_texto,
  sum(length(CAST(unidades AS VARCHAR))
      + length(CAST(satisfaccion AS VARCHAR))) AS bytes_de_numeros
FROM 'ventas-miss-yera.csv'
"""))

┌────────────────┬──────────────────┐
│ bytes_de_texto │ bytes_de_numeros │
│     int128     │      int128      │
├────────────────┼──────────────────┤
│          92401 │             7368 │
└────────────────┴──────────────────┘

chico = os.path.join(carpeta, 'chico.parquet')
con.sql("COPY (SELECT * FROM 'ventas-miss-yera.csv') TO '" + chico
        + "' (FORMAT parquet)")
print('csv    :', os.path.getsize('ventas-miss-yera.csv'), 'bytes')
print('parquet:', os.path.getsize(chico), 'bytes')
print('ahorro :', round((1 - os.path.getsize(chico)
                         / os.path.getsize('ventas-miss-yera.csv')) * 100, 1), '%')

csv    : 278857 bytes
parquet: 88752 bytes
ahorro : 68.2 %

del_grande = [f[0] for f in con.sql(
    "DESCRIBE SELECT * FROM '" + grande + "'").fetchall()]
del_csv = [f[0] for f in con.sql(
    "DESCRIBE SELECT * FROM 'ventas-miss-yera.csv'").fetchall()]
print('en el parquet:', del_grande)
print('se quedaron   :', sorted(set(del_csv) - set(del_grande)))

en el parquet: ['unidades', 'monto', 'canal', 'ciudad', 'repeticion']
se quedaron   : ['categoria', 'cliente_id', 'compro', 'descuento', 'fecha', 'fecha_ultima_compra', 'id_venta', 'monto_final_facturado', 'satisfaccion', 'segmento']

con.sql("SELECT * FROM 'no-existe.parquet'")

IOException: IO Error: No files found that match the pattern "no-existe.parquet"