06 · Scegliere e ordinare le righe

Filtra, ordina e campiona righe d’ordine. La lezione autonoma usa PySpark classico 4.x e Java 17 o successivo. Su Kaggle avvia una nuova sessione CPU. Se serve, installa PySpark con %pip install "pyspark>=4.0,<4.1" in una cella di preparazione, poi usa Run All.

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

spark = SparkSession.builder.master("local[2]").appName("06-choose-and-sort-rows").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
lines = spark.createDataFrame(
    [(1, "tea", 2), (2, "coffee", 5), (3, "iced tea", None),
     (4, "cake", 3), (5, "tea", 5), (6, None, 1)],
    "line_id INT, item STRING, quantity INT",
)
item = F.col("item")
qty = F.col("quantity")

1. Ordinare le righe

orderBy() ordina l’intero risultato. Aggiungi una chiave univoca di spareggio per rendere riproducibile l’ordine tra valori uguali.

2. Condizione

Una condizione è un’espressione di colonna che restituisce vero, falso o sconosciuto. Qui verifica se la quantità è almeno tre.

3. Filtrare le righe

filter() conserva solo le righe per cui la condizione è vera. La quantità sconosciuta viene quindi esclusa.

4. Confrontare valori

Usa espressioni come qty >= 3, qty == 5 e qty != 5. L’uguaglianza usa due segni di uguale.

lines.orderBy("line_id").show()
at_least_three = qty >= 3
assert lines.filter(at_least_three).count() == 3
lines.filter(qty == 5).orderBy("line_id").show()

5. Appartenenza a una lista

isin() verifica se il valore della colonna coincide con uno di quelli forniti.

6. Condizioni sul testo

contains() cerca una sottostringa letterale. startswith() e endswith() verificano l’inizio e la fine della stringa. Gli esempi usano i valori minuscoli dei dati di prova.

7. between include gli estremi

between(2, 5) include sia 2 e 5 sia i valori interni all’intervallo.

assert lines.filter(item.isin("tea", "coffee")).count() == 3
assert lines.filter(item.contains("tea")).count() == 3
assert lines.filter(item.startswith("tea")).count() == 2
assert lines.filter(item.endswith("tea")).count() == 3
assert lines.filter(qty.between(2, 5)).count() == 4

8. Controllare se un valore è null

Usa isNull() per i valori mancanti e isNotNull() per quelli presenti.

9. Confrontare con null

I confronti ordinari con null danno sconosciuto, anche tra due null.

10. Confrontare trattando i null come uguali

eqNullSafe() restituisce vero per due null, falso quando uno solo è null e l’uguaglianza ordinaria per i valori presenti di questo esempio.

11. == None non trova righe

qty == None crea un confronto ordinario con null. Ogni esito è sconosciuto e il filtro conserva zero righe.

12. Il confronto di disuguaglianza esclude i null

qty != 5 mantiene i valori presenti diversi da cinque. Per includere le quantità sconosciute, aggiungi una condizione sui null.

assert lines.filter(qty.isNull()).count() == 1
assert lines.filter(qty.isNotNull()).count() == 5
assert lines.filter(qty.eqNullSafe(None)).count() == 1
assert lines.filter(qty == None).count() == 0
assert lines.filter(qty != 5).count() == 3
assert lines.filter((qty != 5) | qty.isNull()).count() == 4
lines.select("line_id", (qty == 5).alias("equals_five")).orderBy("line_id").show()

13. Combinare condizioni

Usa & per AND, | per OR e ~ per NOT sulle colonne Spark. Negare un esito sconosciuto dà ancora sconosciuto.

14. and, or e not di Python sulle colonne

Gli operatori booleani Python cercano di valutare l’oggetto Column stesso. Spark richiede invece un’espressione da valutare per ogni riga. L’esempio intercetta l’errore e prosegue.

15. Precedenza di & senza parentesi

Python interpreta qty >= 2 & qty <= 5 come un confronto concatenato che coinvolge 2 & qty. Racchiudi ogni confronto tra parentesi.

condition = (qty >= 2) & (qty <= 5)
assert lines.filter(condition).count() == 4
assert lines.filter(~(qty == 5)).count() == 3
try:
    (qty >= 2) and (qty <= 5)
except PySparkValueError as error:
    assert error.getCondition() == "CANNOT_CONVERT_COLUMN_INTO_BOOL"
    print(error.getCondition())
else:
    raise AssertionError("Expected Python boolean conversion to fail.")

16. Campione casuale

sample() usa una frazione. La dimensione effettiva può differire dal prodotto tra frazione e numero di righe. Usa un seme per ripetere il campionamento a parità di input e disposizione delle partizioni.

17. Campione stratificato

sampleBy() accetta una frazione per strato. Qui gli ID pari e dispari formano due strati. Gli strati omessi hanno frazione zero. Le frazioni fissano i tassi di campionamento, mentre i conteggi effettivi possono variare.

18. Dividere le righe in parti casuali

randomSplit() normalizza i pesi relativi forniti. I pesi indicano le proporzioni attese. Controlla i conteggi effettivi.

19. randomSplit senza cache

Azioni separate possono rivalutare l’input comune a monte. Con input e partizioni stabili, un seme fisso è utile. Se il lavoro a monte può cambiare, dividi a partire da uno snapshot stabile. Materializzare la cache può evitare il ricalcolo finché i blocchi rimangono disponibili. Usa uno snapshot persistente per fissare l’input tra tentativi o sessioni.

population = spark.range(200, numPartitions=2).withColumn("stratum", F.col("id") % 2)
print("Sample size:", population.sample(withReplacement=False, fraction=0.2, seed=17).count())
stratified = population.sampleBy("stratum", fractions={0: 0.10, 1: 0.40}, seed=17)
stratified.groupBy("stratum").count().orderBy("stratum").show()

train, test = population.randomSplit([0.8, 0.2], seed=17)
train_ids = {row.id for row in train.select("id").collect()}
test_ids = {row.id for row in test.select("id").collect()}
assert train_ids.isdisjoint(test_ids)
assert train_ids | test_ids == set(range(200))
print("Split sizes:", len(train_ids), len(test_ids))

20. Ordinare in modo decrescente

desc() mette prima i valori presenti più grandi. Usa line_id per risolvere i pari merito sulla quantità.

21. Posizione dei null

Per impostazione predefinita, i null vengono prima nell’ordine crescente e dopo in quello decrescente. Puoi impostare la posizione con asc_nulls_last() o desc_nulls_first().

22. Ordinare dentro le partizioni

sortWithinPartitions() ordina le righe dentro ogni partizione. I confini tra partizioni separano ancora le sequenze ordinate.

23. Ordine delle righe non garantito

Un’anteprima può sembrare stabile tra esecuzioni. Se la logica a valle dipende dall’ordine, applica un ordinamento finale esplicito con chiavi di spareggio.

24. Ordinamento globale con rimescolamento

Un ordinamento globale su più partizioni può redistribuire le righe in intervalli ordinati. Nel piano dell’esempio cerca uno scambio con partizionamento per intervalli e un ordinamento.

descending = lines.orderBy(qty.desc(), "line_id")
assert [row.line_id for row in descending.collect()] == [2, 5, 4, 1, 6, 3]
null_first = lines.orderBy(qty.desc_nulls_first(), "line_id")
assert null_first.first().line_id == 3
lines.orderBy(qty.asc_nulls_last(), "line_id").show()

local_sort = population.repartition(2, "stratum").sortWithinPartitions(F.col("id").desc())
local_sort.explain()
population.orderBy(F.col("stratum").desc(), F.col("id").desc()).explain()

25. Campionare esattamente n righe

Su una tabella piccola, assegna a ogni riga una chiave pseudocasuale con rand(seed), ordina per chiave e prendi le prime n righe. Ottieni n righe se sono disponibili. Un ID univoco risolve i pari merito. I numeri casuali tornano nel Notebook 07. Il metodo valuta una chiave su tutto l’input e può essere costoso su larga scala.

exact_sample = population.orderBy(F.rand(17), "id").limit(5)
assert exact_sample.count() == 5
exact_sample.show()

Verifica la comprensione

1. Quale espressione conserva le quantità sconosciute e quelle diverse da cinque?
2. Perché un campione al 20% può avere un numero di righe diverso dal 20% dell’input?
3. Quale ordinamento serve per ottenere un ordine unico tra tutte le partizioni?

Risposte

1. (qty != 5) | qty.isNull().
2. La frazione determina un tasso di campionamento. La dimensione effettiva varia.
3. Usa orderBy() e aggiungi chiavi di spareggio quando conta l’ordine tra pari merito.