15 · Leggere dati

Leggi piccoli file, controlla gli schemi ed esamina i record scartati. Esegui in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se necessario, installa PySpark con %pip install "pyspark[sql]>=4.0,<4.1". Il notebook crea i propri file. L’esempio JDBC richiede un database esterno e si esegue solo se lo abiliti esplicitamente.

import os
from pathlib import Path
from tempfile import TemporaryDirectory

from pyspark.errors import PySparkException
from py4j.protocol import Py4JJavaError
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition

spark = SparkSession.builder.master("local[2]").appName("15-read-data").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
workspace = TemporaryDirectory(prefix="spark-nb15-")
root = Path(workspace.name)

def fixture(name, content):
    path = root / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    return str(path)

csv_path = fixture("orders.csv", "id,day,units\n1,2026-11-01,2\n2,2026-11-02,3\n")
json_path = fixture("orders.json", '{"id":1,"units":2}\n{"id":2,"units":3}\n')

1. Leggere un file CSV in una tabella

spark.read.csv(path) legge record delimitati in un DataFrame.

2. Usare la prima riga per i nomi

Imposta header=True quando la prima riga contiene i nomi delle colonne. Altrimenti Spark assegna nomi come _c0 e tratta la prima riga come dato.

3. Dedurre i tipi in lettura

Per impostazione predefinita, i campi CSV sono stringhe. inferSchema=True chiede a Spark di esaminare i valori e dedurne i tipi, il che può richiedere un passaggio in più.

4. Leggere JSON

spark.read.json() legge per impostazione predefinita JSON delimitato da nuove righe. Qui ogni riga è un oggetto JSON distinto.

assert spark.read.csv(csv_path).count() == 3
strings = spark.read.csv(csv_path, header=True)
inferred = spark.read.csv(csv_path, header=True, inferSchema=True)
assert strings.schema["units"].dataType.simpleString() == "string"
assert inferred.schema["units"].dataType.simpleString() == "int"
spark.read.json(json_path).orderBy("id").show()

5. Sorgente di dati

Una sorgente indica a Spark come trasformare l’input esterno in righe. CSV, JSON, Parquet e JDBC sono alcuni esempi.

6. Forma generale della lettura

Usa spark.read.format(...).options(...).load(...) per specificare la sorgente e le sue opzioni.

7. Estensione e separatore del file

Un separatore divide i campi. Nel prossimo file è un punto e virgola. L’estensione è una convenzione sul nome del file.

8. Imporre lo schema in lettura

Passa uno schema quando conosci i tipi dei campi. Scegli opzioni compatibili con il contratto della sorgente e verifica l’intestazione e i dati letti.

9. Quando la deduzione lascia le date come testo

Spark può dedurre DATE da input adatti. Una colonna che sembra contenere date può restare STRING se la deduzione incontra testo incompatibile o se la preferenza per le date è disattivata. Controlla lo schema prima di calcolare con le date.

10. Formato delle date in lettura

Quando un campo dichiarato DATE usa un formato personalizzato, imposta dateFormat di conseguenza. Qui dd/MM/yyyy indica giorno, mese e anno di calendario.

semicolon = fixture("delivery.csv", "id;day;units\n1;01/11/2026;2\n")
typed = (
    spark.read.format("csv").options(header=True, sep=";", dateFormat="dd/MM/yyyy")
    .schema("id INT, day DATE, units INT").load(semicolon)
)
assert str(typed.first().day) == "2026-11-01"
as_text = spark.read.options(header=True, inferSchema=True, preferDate=False).csv(csv_path)
assert as_text.schema["day"].dataType.simpleString() == "string"

11. Leggere un file come testo

spark.read.text() restituisce una colonna stringa chiamata value, con una riga per linea di testo per impostazione predefinita.

12. Opzioni di forma del CSV

Opzioni come sep, quote e multiLine descrivono come sono codificati i record. L’esempio usa un punto e virgola in un campo tra virgolette.

13. L’estensione .csv non identifica il separatore

Un file .csv può contenere campi separati da virgole o punti e virgola. Usa il contratto del produttore o esamina un campione del testo.

14. Separatore sbagliato

Un lettore impostato sulla virgola vede un solo campo nell’intestazione separata da punti e virgola. Verifica nomi e tipi delle colonne risultanti.

spark.read.text(semicolon).show(truncate=False)
quoted = fixture("quoted.csv", 'id;note\n1;"tea; no sugar"\n')
assert spark.read.options(header=True, sep=";", quote='"', multiLine=True).csv(quoted).first().note == "tea; no sugar"
assert spark.read.csv(semicolon, header=True).columns == ["id;day;units"]

15. Modalità per i dati non validi

PERMISSIVE conserva i campi recuperabili e può catturare il record errato originale. DROPMALFORMED scarta i record malformati quando li rileva. FAILFAST genera un errore quando ne incontra uno.

16. Conservare il record non valido

Dichiara un campo stringa chiamato _corrupt_record e usalo come colonna dei record corrotti. I dati di prova contengono un errore di conversione di tipo.

17. Perché count() può non rilevare righe malformate

Un conteggio può escludere campi inutilizzati ed evitare di interpretare il valore non valido. Da solo è un’azione di convalida inaffidabile. Confrontalo con la materializzazione di tutti i campi nella modalità scelta.

18. Interrogare solo il campo dei record corrotti

Per sorgenti CSV e JSON grezze, Spark limita le query che fanno riferimento soltanto al campo dei record corrotti. Metti in cache e materializza il DataFrame letto prima di proiettare quel solo campo.

bad_path = fixture("bad.csv", "id,units\n1,2\n2,wrong\n3,4\n")

def read_bad(mode):
    return spark.read.options(
        header=True, mode=mode, columnNameOfCorruptRecord="_corrupt_record",
    ).schema("id INT, units INT, _corrupt_record STRING").csv(bad_path)

dropped = read_bad("DROPMALFORMED")
print("Count with possible pruning:", dropped.count())
assert {r.id for r in dropped.collect()} == {1, 3}
try:
    read_bad("FAILFAST").collect()
except (PySparkException, Py4JJavaError) as error:
    assert "MALFORMED_RECORD_IN_PARSING" in str(error)
    print("FAILFAST rejected the malformed record.")
else:
    raise AssertionError("Expected an invalid integer to fail.")

parsed = read_bad("PERMISSIVE").cache()
try:
    parsed.collect()
    rejected = parsed.select("_corrupt_record").where("_corrupt_record IS NOT NULL").collect()
    assert [row[0] for row in rejected] == ["2,wrong"]
finally:
    parsed.unpersist()

19. Leggere una cartella o un modello di nomi

Passa una cartella per un dataset con file compatibili. Usa pathGlobFilter per limitare i nomi dei file senza alterare la scoperta delle partizioni.

20. Scoperta delle partizioni

Spark riconosce i nomi delle cartelle con coppie chiave-valore, come zone=A. Leggi la radice del dataset o fornisci basePath quando leggi solo alcune sottocartelle.

21. Colonna nascosta dei metadati dei file

Alcune sorgenti file Spark espongono una struct nascosta _metadata. Se supportato, selezionane un campo esplicitamente e controlla che nessuna colonna di input nasconda il nome dei metadati. La lezione usa input_file_name() di Apache Spark per mostrare la provenienza del file in Spark classico.

fixture("zones/zone=A/a.csv", "id,units\n1,2\n")
fixture("zones/zone=B/b.csv", "id,units\n2,3\n")
fixture("zones/zone=B/notes.txt", "ignore this")
zones = spark.read.options(header=True, pathGlobFilter="*.csv").schema("id INT, units INT").csv(str(root / "zones"))
assert {r.zone for r in zones.select("zone").collect()} == {"A", "B"}
zones.select("*", F.input_file_name().alias("source_file")).show(truncate=False)

22. Leggere Parquet

Parquet conserva i tipi insieme ai dati. Scrivi un piccolo dataset locale e rileggilo. Qui coalesce(1) riduce le partizioni di output senza richiedere uno shuffle.

23. Unire gli schemi dei file

Usa mergeSchema=true per leggere insieme varianti compatibili dello schema. Il campo aggiunto vale null nelle righe più vecchie.

24. mergeSchema con molti file

Unire gli schemi richiede l’ispezione dei metadati dei file e può aumentare il costo di pianificazione. Con input di produzione stabili, gestisci esplicitamente l’evoluzione e fornisci uno schema noto quando opportuno.

25. Leggere un file nuovo

Crea una nuova lettura dopo aver aggiunto file. DataFrame e indici dei file possono riflettere quelli individuati durante la pianificazione. Ricrea o aggiorna i lettori quando cambia la sorgente.

parquet_path = str(root / "parquet")
spark.createDataFrame([(1, 2)], "id INT, units INT").coalesce(1).write.parquet(parquet_path)
assert spark.read.parquet(parquet_path).schema["units"].dataType.simpleString() == "int"
spark.createDataFrame([(2, 3, "web")], "id INT, units INT, channel STRING").coalesce(1).write.mode("append").parquet(parquet_path)
merged = spark.read.option("mergeSchema", True).parquet(parquet_path)
assert {r.id: r.channel for r in merged.collect()} == {1: None, 2: "web"}

26. Leggere da un database

Il lettore JDBC usa un URL JDBC, una tabella e un driver compatibile sul classpath di Spark. Imposta RUN_JDBC=1 solo dopo aver configurato database, driver e variabili d’ambiente indicate. Per impostazione predefinita Run All salta la chiamata esterna.

if os.environ.get("RUN_JDBC") == "1":
    database = spark.read.format("jdbc").options(
        url=os.environ["DEMO_JDBC_URL"],
        dbtable=os.environ["DEMO_JDBC_TABLE"],
        driver=os.environ["DEMO_JDBC_DRIVER"],
        user=os.environ["DEMO_JDBC_USER"],
        password=os.environ["DEMO_JDBC_PASSWORD"],
    ).load()
    database.printSchema()
else:
    print("JDBC example skipped. Set RUN_JDBC=1 after configuring the database and driver.")

27. Sorgente dati scritta in Python

Spark 4 può registrare una sorgente implementata in Python. Il lettore fornisce partizioni di input ed emette righe dello schema dichiarato. L’esempio usa due descrittori di partizione e funziona senza servizi esterni.

class DemoReader(DataSourceReader):
    def partitions(self):
        return [InputPartition(1), InputPartition(2)]

    def read(self, partition):
        yield (partition.value, partition.value * 10)

class DemoSource(DataSource):
    @classmethod
    def name(cls):
        return "nb15_demo"

    def schema(self):
        return "id INT, units INT"

    def reader(self, schema):
        return DemoReader()

spark.dataSource.register(DemoSource)
demo = spark.read.format("nb15_demo").load()
assert [tuple(r) for r in demo.orderBy("id").collect()] == [(1, 10), (2, 20)]
demo.show()
workspace.cleanup()

Verifica la comprensione

1. Perché un file .csv può produrre una sola colonna?
2. Perché il solo conteggio delle righe è insufficiente per controllare input malformati?
3. Che cosa succede a una colonna aggiunta nei file Parquet più recenti quando si uniscono gli schemi?

Risposte

1. Il separatore configurato può essere diverso da quello del file.
2. Spark può escludere i campi che il conteggio non usa.
3. La nuova colonna viene inclusa e vale null per le righe più vecchie che non avevano quel campo.