18 · Usare SQL

Interroga le stesse vendite del bar con SQL e con l’API DataFrame. 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", poi usa Run All. Tutti i dati sono nel notebook.

1. Spark SQL

Spark SQL interroga dati strutturati. Qui ogni riga rappresenta una vendita e i prezzi sono espressi in centesimi interi.

2. Registrare una vista temporanea

createOrReplaceTempView() assegna a un DataFrame un nome SQL nella sessione corrente.

3. Eseguire una query SQL

spark.sql() restituisce un DataFrame per una SELECT. Usa show() per mostrare il risultato e ORDER BY quando conta l’ordine.

from uuid import uuid4

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

spark = SparkSession.builder.master("local[2]").appName("18-use-sql").getOrCreate()
sales = spark.createDataFrame(
    [(101, "tea", 2, 250), (102, "coffee", 1, 375), (103, "tea", 4, 250), (104, None, 3, 100)],
    "sale_id INT, item STRING, units INT, price_cents INT",
)
view = "nb18_sales_" + uuid4().hex[:8]
sales.createOrReplaceTempView(view)
spark.sql(f"SELECT sale_id, item, units FROM {view} ORDER BY sale_id").show()

4. Un solo motore per SQL e DataFrame

Entrambe le interfacce usano il motore di query Spark. Usa explain() per esaminare i piani. Query equivalenti possono mostrare dettagli diversi, come gli ID delle espressioni.

5. Selezionare con un’espressione SQL

selectExpr() accetta un’espressione SQL per colonna di output. Usa AS per nominare i campi calcolati.

6. Scrivere una condizione come testo SQL

filter() accetta un predicato SQL come stringa o un’espressione Column. Gli esempi usano predicati fissi. Per valori esterni nelle query usa parametri.

7. Condizione in SQL

Usa WHERE per conservare le righe per cui il predicato è vero. Un confronto con un prodotto null dà sconosciuto, quindi la riga non passa.

8. Espressione SQL in una colonna

expr() interpreta un’espressione SQL come Column. Usala dentro metodi come withColumn().

query = spark.sql(f'''
    SELECT sale_id, units * price_cents AS revenue_cents
    FROM {view}
    WHERE item = 'tea' AND units >= 2
''')
selected = sales.filter("item = 'tea' AND units >= 2").selectExpr(
    "sale_id", "units * price_cents AS revenue_cents"
)
computed = (
    sales.filter((F.col("item") == "tea") & (F.col("units") >= 2))
    .withColumn("revenue_cents", F.expr("units * price_cents"))
    .select("sale_id", "revenue_cents")
)
query.explain()
selected.explain()
query.orderBy("sale_id").show()

9. Vista temporanea legata alla sessione

Una nuova sessione Spark può condividere lo SparkContext e avere viste temporanee locali separate. Verifica la visibilità in entrambe le sessioni.

10. Controllare se due query sono equivalenti

Confronta gli schemi, allinea le colonne e usa exceptAll() in entrambe le direzioni per confrontare le righe, compresi i conteggi dei duplicati. Il controllo vale per questi dati di prova e non dimostra l’equivalenza su tutti gli input.

assert spark.catalog.tableExists(view)
assert not spark.newSession().catalog.tableExists(view)

def check_same_rows(left, right):
    assert left.schema == right.schema
    assert left.exceptAll(right).isEmpty()
    assert right.exceptAll(left).isEmpty()

check_same_rows(query, selected)
check_same_rows(query, computed)
assert {r.sale_id: r.revenue_cents for r in query.collect()} == {101: 500, 103: 1000}

11. Parametri con nome in una query

Scrivi un segnaposto con nome come :item_name e passa il valore nel dizionario args. Il valore viene associato senza essere inserito nel testo della query.

12. DataFrame come argomento di una query

L’argomento con nome in spark.sql(..., source=sales) permette a Spark di inserire un riferimento al DataFrame. Usa un segnaposto come {source} dove serve una relazione. I segnaposto con nome continuano ad associare i valori separatamente.

13. Iniezione SQL

Inserire valori grezzi dell’utente nel testo SQL permette loro di alterare la struttura della query. Associa i valori con segnaposto. Per identificativi selezionabili dall’utente, risolvili usando un elenco di identificativi ammessi. Qui si associa testo con apici e caratteri simili a SQL e si verifica che non corrisponda ad alcun prodotto.

params = {"item_name": "tea", "minimum_units": 2}
sql_template = '''
    SELECT sale_id, units * price_cents AS revenue_cents
    FROM {source}
    WHERE item = :item_name AND units >= :minimum_units
'''
parameterized = spark.sql(sql_template, args=params, source=sales)
check_same_rows(query, parameterized)
literal_only = spark.sql(
    sql_template,
    args={"item_name": "tea' OR 1=1 --", "minimum_units": 2},
    source=sales,
)
assert literal_only.isEmpty()

14. Sintassi SQL a pipe

Spark 4 supporta gli operatori SQL a pipe. Parti da una relazione e applica le fasi nell’ordine scritto con |>. La query filtra le vendite, seleziona campi calcolati e ordina l’output.

piped = spark.sql(f'''
    FROM {view}
    |> WHERE item = 'tea' AND units >= 2
    |> SELECT sale_id, units * price_cents AS revenue_cents
    |> ORDER BY sale_id
''')
check_same_rows(query, piped)
assert [r.sale_id for r in piped.collect()] == [101, 103]
piped.show()
spark.catalog.dropTempView(view)

Verifica la comprensione

1. Quale API trasforma un’espressione SQL testuale in una Column?
2. Perché confrontare i risultati in entrambe le direzioni con exceptAll?
3. Come dovresti fornire un nome di prodotto proveniente dall’esterno?

Risposte

1. Usa F.expr().
2. Ogni direzione rileva righe o occorrenze duplicate presenti da un solo lato.
3. Associalo come valore con un parametro con nome.