16 · Scrivere dati

Scrivi file e tabelle di catalogo, poi verifica il risultato. Usa una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa con %pip install "pyspark>=4.0,<4.1". La lezione crea destinazioni temporanee.

from pathlib import Path
from tempfile import TemporaryDirectory
from uuid import uuid4

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

workspace = TemporaryDirectory(prefix="spark-nb16-")
root = Path(workspace.name)
spark = (
    SparkSession.builder.master("local[2]").appName("16-write-data")
    .config("spark.sql.catalogImplementation", "in-memory")
    .config("spark.sql.warehouse.dir", (root / "warehouse").as_uri())
    .getOrCreate()
)
spark.conf.set("spark.sql.shuffle.partitions", "2")
sales = spark.createDataFrame(
    [(101, 2, "North"), (102, 3, "South"), (103, 1, "North"), (104, 4, "South")],
    "order_id INT, units INT, shop STRING",
)

1. Catalogo

Il catalogo permette a Spark di risolvere i nomi delle tabelle e consultare metadati come schema e posizione. Puoi leggere un percorso direttamente, senza una voce di catalogo. Qui si usa un catalogo in memoria locale alla sessione.

database = "nb16_" + uuid4().hex[:12]
spark.sql(f"CREATE DATABASE {database} LOCATION '{(root / database).as_uri()}'")

2. Scrivere in Parquet

Usa df.write.parquet(path) per scrivere un dataset Parquet. La destinazione è di solito una directory con file di dati.

3. Ridurre le partizioni senza rimescolare

coalesce(1) riduce le partizioni Spark senza aggiungere uno shuffle. Qui i dati sono pochi. Un unico task di scrittura può rallentare output grandi.

parquet = str(root / "sales_parquet")
sales.coalesce(1).write.parquet(parquet)
assert spark.read.parquet(parquet).count() == 4

4. Modalità di salvataggio

La modalità predefinita errorifexists genera un errore se la destinazione esiste. ignore la salta, append aggiunge dati e overwrite la sostituisce secondo le regole della sorgente.

5. Forma generale della scrittura

Usa write.format(...).mode(...).options(...).save(path) per specificare formato e opzioni.

batch_path = str(root / "batches")
sales.limit(1).write.format("parquet").mode("errorifexists").save(batch_path)
sales.write.format("parquet").mode("ignore").save(batch_path)
assert spark.read.parquet(batch_path).count() == 1
sales.write.format("parquet").mode("append").save(batch_path)
assert spark.read.parquet(batch_path).count() == 5

6. Formati per righe e per colonne

Il CSV codifica record completi come righe di testo. Parquet memorizza i valori per colonna all’interno di gruppi di righe. I lettori compatibili possono così saltare le colonne inutili.

7. Il CSV non conserva i tipi

Il CSV non incorpora lo schema Spark. In lettura, fornisci uno schema o deducilo. In assenza di entrambi, l’ID numerico viene letto come stringa.

8. File e partizioni in memoria

In una scrittura semplice senza partizionamento su disco, ogni task con dati scrive di norma un file. Partizioni vuote, limiti di record e partizionamento in directory possono cambiare il conteggio. Ignora i file marcatori nell’ispezione.

csv_path = str(root / "sales_csv")
sales.coalesce(1).write.option("header", True).csv(csv_path)
assert spark.read.option("header", True).csv(csv_path).schema["order_id"].dataType.simpleString() == "string"
assert len(list(Path(parquet).glob("part-*.parquet"))) == 1

9. Partizionare i file in directory

partitionBy("shop") colloca le righe in directory come shop=North. Queste partizioni su disco sono distinte dalle partizioni Spark in memoria.

10. Overwrite può sostituire tutta la destinazione

Questa sovrascrittura Parquet senza partizionamento sostituisce l’intero dataset temporaneo. Le quattro righe originali lasciano il posto alla sola riga selezionata. Verifica il percorso prima di sovrascrivere.

11. Righe massime per file

maxRecordsPerFile può suddividere l’output di un task in più file. Limita il numero di righe, indipendentemente dai byte.

partitioned = str(root / "by_shop")
sales.coalesce(1).write.partitionBy("shop").parquet(partitioned)
assert (Path(partitioned) / "shop=North").is_dir()

sales.filter("order_id = 101").write.mode("overwrite").parquet(parquet)
assert [r.order_id for r in spark.read.parquet(parquet).collect()] == [101]

limited = root / "limited"
sales.coalesce(1).write.option("maxRecordsPerFile", 2).parquet(str(limited))
assert len(list(limited.glob("part-*.parquet"))) == 2

12. Potatura delle partizioni

Un filtro sulla chiave di partizione su disco può permettere a Spark di saltare altre directory. Cerca il filtro di partizione nel piano di lettura.

13. File piccoli

Molti file piccoli aumentano il costo di elencarli, aprirli e gestire i task. Controlla conteggi e dimensioni quando scegli il numero di task di scrittura.

14. Ripartire per colonna

repartition(2, "shop") rimescola le righe per collocare le chiavi di negozio uguali nella stessa partizione Spark. Combinalo con il partizionamento in directory quando serve quella disposizione.

15. Regole per la colonna di partizione

Scegli una colonna usata spesso nei filtri, con cardinalità gestibile e dati sufficienti per valore. Un ID ordine univoco creerebbe una directory per ordine. Considera valori mancanti e chiavi sbilanciate.

north = spark.read.parquet(partitioned).filter(F.col("shop") == "North")
assert north.count() == 2
north.explain()
print([(file.name, file.stat().st_size) for file in sorted(limited.glob("part-*.parquet"))])
sales.repartition(2, "shop").write.partitionBy("shop").parquet(str(root / "clustered"))

16. Salvare una tabella nel catalogo

saveAsTable() scrive i dati e registra una tabella. Senza percorso esplicito, qui usa la posizione gestita del database.

17. Elencare le tabelle del catalogo

spark.catalog.listTables(database) elenca le tabelle visibili nel database.

18. Leggere una tabella per nome

Usa spark.table() con il nome qualificato. Spark consulta il catalogo per trovare i dati.

managed = f"{database}.sales"
sales.write.format("parquet").saveAsTable(managed)
assert "sales" in {table.name for table in spark.catalog.listTables(database)}
assert spark.table(managed).count() == 4

19. Inserire righe in una tabella

insertInto() scrive in una tabella esistente. Qui accoda una riga.

20. insertInto abbina per posizione

Lo schema di destinazione è già fissato. Spark assegna i campi in ingresso per posizione. Due interi scambiati possono superare i controlli di tipo e finire nei campi sbagliati. L’esempio temporaneo mostra l’errore, poi allinea le colonne prima del secondo inserimento.

extra = spark.createDataFrame([(105, 5, "North")], sales.schema)
swapped = extra.select("units", "order_id", "shop")
swapped.write.insertInto(managed)
bad = spark.table(managed).filter("order_id = 5").first()
assert bad.units == 105
swapped.select(*spark.table(managed).columns).write.insertInto(managed)
assert spark.table(managed).filter("order_id = 105").first().units == 5

21. Tabella con percorso esplicito

Registra la tabella indicando il percorso dei dati. In questo esempio i file rimangono dopo aver eliminato la tabella esterna.

22. Creare una tabella con SQL

CREATE TABLE ... USING parquet dichiara una tabella di catalogo. Qui una tabella gestita di prova riceve una riga via SQL.

23. Svuotare una tabella

TRUNCATE TABLE rimuove le righe da una tabella gestita compatibile e ne conserva la definizione. L’esempio verifica entrambe le proprietà prima di rimuovere il database temporaneo.

external = f"{database}.external_sales"
external_path = root / "external_sales"
sales.write.format("parquet").option("path", external_path.as_uri()).saveAsTable(external)
spark.sql(f"DROP TABLE {external}")
assert spark.read.parquet(str(external_path)).count() == 4

sql_table = f"{database}.sql_sales"
spark.sql(f"CREATE TABLE {sql_table} (order_id INT, units INT) USING parquet")
spark.sql(f"INSERT INTO {sql_table} VALUES (101, 2)")
assert spark.table(sql_table).count() == 1
spark.sql(f"TRUNCATE TABLE {sql_table}")
assert spark.table(sql_table).count() == 0
assert spark.table(sql_table).columns == ["order_id", "units"]
spark.sql(f"DROP DATABASE {database} CASCADE")
workspace.cleanup()

Verifica la comprensione

1. Quale operazione crea le directory di partizione?
2. Perché insertInto può accettare colonne intere scambiate?
3. Che cosa conserva TRUNCATE?

Risposte

1. DataFrameWriter.partitionBy() crea directory dai valori delle chiavi.
2. Abbina i campi per posizione e i valori sono compatibili con i tipi di destinazione.
3. La definizione della tabella nel catalogo, incluso lo schema.