22 · Configurare ed eseguire un’applicazione Spark

Distingui le risorse di avvio dalle impostazioni SQL di sessione, poi esegui un’applicazione autonoma. Usa PySpark classico 4.0.x e Java 17+. In una nuova sessione CPU su Kaggle, installa PySpark se serve con %pip install "pyspark>=4.0,<4.1", poi usa Run All. Il sottoprocesso locale non richiede dati esterni.

1. getOrCreate con una nuova configurazione

getOrCreate() può restituire la sessione esistente. In tal caso, le impostazioni di sessione modificabili passate al builder possono avere effetto. Lo SparkContext conserva le risorse definite all’avvio.

2. SparkContext

Lo SparkContext collega l’applicazione all’ambiente di esecuzione. PySpark classico lo espone come spark.sparkContext.

3. SparkConf

SparkConf descrive le proprietà dell’applicazione. getConf() restituisce una copia della configurazione del contesto.

4. Leggere e cambiare un’impostazione

Usa spark.conf.get() e set() per la configurazione SQL durante l’esecuzione.

5. Impostazione modificabile in sessione

Usa isModifiable() per verificare se RuntimeConfig può cambiare un’impostazione. Cambiare il numero di partizioni di shuffle può influire sui piani SQL futuri.

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

launch = SparkConf().setMaster("local[2]").setAppName("22-configure-and-run")
spark = SparkSession.builder.config(conf=launch).getOrCreate()
sc = spark.sparkContext
key = "spark.sql.shuffle.partitions"
original = spark.conf.get(key)
try:
    same = SparkSession.builder.config(key, "5").getOrCreate()
    assert same.sparkContext is sc
    assert same.conf.get(key) == "5"
    assert spark.conf.isModifiable(key)
    spark.conf.set(key, "3")
    assert spark.conf.get(key) == "3"
    assert not spark.conf.isModifiable("spark.executor.memory")
finally:
    spark.conf.set(key, original)

print("Application:", sc.appName, "Master:", sc.getConf().get("spark.master"))

6. spark-submit

spark-submit avvia un’applicazione Spark con opzioni di lancio seguite dal programma e dai suoi argomenti. L’esempio scrive un piccolo programma in una directory temporanea e lo esegue in un processo separato.

7. Valori di --master

local[2] usa due thread locali. Nei cluster puoi trovare un URL Standalone come spark://host:7077, yarn o un URL dell’API Kubernetes con prefisso k8s://. Usa il valore adatto al gestore installato.

8. Modalità di deploy

In modalità client il driver parte nell’ambiente che invia l’applicazione. In modalità cluster il driver parte nel cluster, se gestore e linguaggio lo supportano. La dimostrazione usa la modalità client.

9. Notebook in modalità client

In questo notebook classico il driver Python rimane con il kernel, quindi chiudere il kernel può interrompere l’applicazione. Lo script inviato ha un proprio ciclo di vita del driver.

import os
from pathlib import Path
import shutil
import subprocess
import sys
from tempfile import TemporaryDirectory

import pyspark

script = '''
import sys
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("nb22-standalone").getOrCreate()
try:
    limit = int(sys.argv[1])
    partitions = spark.conf.get("spark.sql.shuffle.partitions")
    print(f"NB22_COUNT={spark.range(limit).count()} PARTITIONS={partitions}")
finally:
    spark.stop()
'''
submit = shutil.which("spark-submit") or str(Path(pyspark.__file__).parent / "bin" / "spark-submit")
with TemporaryDirectory(prefix="nb22-") as directory:
    app = Path(directory) / "count_rows.py"
    app.write_text(script, encoding="utf-8")
    environment = dict(os.environ, PYSPARK_PYTHON=sys.executable)
    completed = subprocess.run(
        [submit, "--master", "local[2]", "--deploy-mode", "client",
         "--driver-memory", "512m", "--conf", "spark.ui.enabled=false",
         "--conf", "spark.sql.shuffle.partitions=3", str(app), "7"],
        env=environment, capture_output=True, text=True, timeout=180,
    )
    if completed.returncode:
        raise RuntimeError(completed.stderr[-4000:])
    assert "NB22_COUNT=7 PARTITIONS=3" in completed.stdout
    print(completed.stdout.strip())

10. Risorse di un executor

In un cluster, i core dell’executor limitano la concorrenza dei task secondo le risorse richieste. La memoria dell’executor determina l’heap JVM. In locale il lavoro condivide una macchina e non ci sono executor remoti dimensionati separatamente.

11. Memoria extra dell’executor

Nei gestori compatibili, l’overhead copre la memoria fuori heap e altre esigenze dei processi. In genere include i worker Python quando la memoria PySpark non ha un’allocazione separata.

12. Allocazione dinamica

L’allocazione dinamica varia il numero di executor in base al carico entro limiti configurati. Richiede un meccanismo supportato per conservare o recuperare lo shuffle, come il tracciamento dello shuffle. L’heap di un executor già avviato mantiene la propria dimensione.

13. Heap più overhead oltre la memoria del nodo

Un heap di 12 GiB e 2 GiB di overhead richiedono almeno 14 GiB, prima delle allocazioni separate per PySpark o fuori heap. Due executor di questo tipo non entrano in 24 GiB di memoria allocabile.

14. Dimensionare gli executor

Fai rientrare memoria e core nelle risorse allocabili del nodo. Verifica poi durata dei task, spill, garbage collection e memoria dei processi. Il calcolo è illustrativo e non richiede risorse del cluster.

node_memory_gib, node_cores = 24, 8
heap_gib, overhead_gib, executor_cores = 12, 2, 4
memory_slots = node_memory_gib // (heap_gib + overhead_gib)
cpu_slots = node_cores // executor_cores
assert min(memory_slots, cpu_slots) == 1
print("Executors per node under these assumptions:", min(memory_slots, cpu_slots))

15. Livelli di configurazione

Le impostazioni possono provenire da SparkConf nel codice, opzioni di invio e file predefiniti. In genere SparkConf prevale se la proprietà può ancora essere applicata. Le proprietà di deploy possono richiedere un’impostazione prima dell’avvio. Le impostazioni SQL di sessione possono poi governare la pianificazione delle query.

16. Impostazioni delle risorse durante l’esecuzione

Imposta heap del driver, heap degli executor e risorse simili all’avvio. Un’opzione SQL su una sessione esistente non ridimensiona le JVM. L’esempio autonomo passa la memoria del driver al programma di avvio.

17. Credenziali da variabili d’ambiente

Leggi le credenziali da una variabile d’ambiente nominata o dal gestore dei segreti della piattaforma. Evita di stampare i valori. Driver ed executor remoti possono avere ambienti diversi. Qui si controlla solo la presenza di una variabile dimostrativa.

18. Autenticazione e cifratura

L’autenticazione verifica chi può connettersi. La cifratura protegge i dati in transito o a riposo, secondo la configurazione. RPC Spark, interfaccia web e sorgenti esterne hanno controlli separati. Configurali secondo le impostazioni approvate per il deployment.

19. Configurazioni visibili

Quando accessibile, la scheda Environment mostra le proprietà configurate esplicitamente. Nel codice ispeziona una lista di proprietà innocue senza stampare l’intera configurazione. I valori predefiniti possono essere assenti da SparkConf.

print("Demo credential configured:", bool(os.environ.get("NB22_DEMO_TOKEN")))
for name in ("spark.master", "spark.app.name"):
    print(name, sc.getConf().get(name))
print("SQL shuffle partitions:", spark.conf.get(key))
print("Spark UI:", sc.uiWebUrl)

20. Novità di Spark 4.x

Spark 4.0 abilita per impostazione predefinita la modalità ANSI SQL e aggiunge API come quelle per leggere VARIANT. Consulta la guida di migrazione per le versioni esatte di partenza e destinazione.

21. Migrare a Spark 4 con dati sporchi

Una conversione che prima tollerava testo malformato può fallire in modalità ANSI. Prova valori errati rappresentativi, usa conversioni sicure dove opportuno e segnala quelle fallite. I dati di prova distinguono testo malformato e valori mancanti.

from pyspark.errors import PySparkException

raw = spark.createDataFrame([("12",), ("bad",), (None,)], "raw_units STRING")
ansi = "spark.sql.ansi.enabled"
prior_ansi = spark.conf.get(ansi)
try:
    spark.conf.set(ansi, "true")
    try:
        raw.select(F.col("raw_units").cast("int")).collect()
    except PySparkException as error:
        assert error.getCondition() == "CAST_INVALID_INPUT"
        print(error.getCondition())
    else:
        raise AssertionError("Expected malformed text to fail.")

    typed = raw.withColumn("units", F.col("raw_units").try_cast("int"))
    checked = typed.withColumn("invalid", F.col("raw_units").isNotNull() & F.col("units").isNull())
    assert {r.raw_units: (r.units, r.invalid) for r in checked.collect()} == {
        "12": (12, False), "bad": (None, True), None: (None, False)
    }
    checked.show()
finally:
    spark.conf.set(ansi, prior_ansi)

Verifica la comprensione

1. Dove imposti l’heap del driver per l’avvio autonomo in modalità client?
2. Che cosa cambia l’allocazione dinamica?
3. Come distingui una quantità mancante da una conversione fallita?

Risposte

1. Passa --driver-memory a spark-submit.
2. Il numero di executor entro i limiti configurati.
3. Conserva l’input originale e segnala i valori presenti la cui conversione sicura dà null.