Avvia letture di dati JDBC simultanee

Hai mai seguito il processo di implementazione di Spark nel tuo progetto, determinando il numero ottimale di partizioni shuffle, l'allocazione di memoria per le istanze di driver ed executor, il numero di core executor e tutte quelle cose divertenti solo per leggere i dati da una fonte JDBC come questa ?

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("query", query) \
    .load()

Qual è il problema con il codice sopra? Bene ... è molto lento :) Se esegui una query SQL in Spark come nell'esempio precedente, utilizzerai solo un thread (puoi vedere nell'interfaccia utente di Spark che viene eseguita solo un'attività durante l'esecuzione del codice) e il dataframe creato lo farà avere una partizione.

La maggior parte di noi utilizza Spark nelle pipeline ETL/ELT per l'elaborazione parallela, quindi leggere i dati da un'origine JDBC utilizzando un solo thread probabilmente non è quello che vogliamo. Per non parlare del fatto che dovremo ripartire il dataframe dopo aver eseguito la query perché i dati sono ora archiviati in 1 partizione. Quindi, come leggiamo effettivamente i dati contemporaneamente in Spark?

# pass the original query to dbtable as a subquery 
final_query = f'({query}) as q'

# depends on your usecase, how many executor cores you have in the cluster
# what do you plan to do with the dataframe downstream, data size etc
partitionsCount = 100

# I'm hardcoding the date values here, but I'll show below how to identify
# these values at run time if you don't know them upfront
min_date = '2020-01-01'
max_date = '2022-12-31'

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("dbtable", final_query) \
    .option("numPartitions", partitionsCount) \
    .option("partitionColumn", "date") \
    .option("lowerBound", f"{min_date}") \
    .option("upperBound", f"{max_date}") \
    .load()

1. dbtable:
   in base alla documentazione ufficiale di Spark, se vogliamo leggere i dati contemporaneamente e partizionarli in base alle nostre esigenze, non possiamo più utilizzare l' opzione query e dobbiamo invece passare la nostra query SQL tramite l' opzione dbtable . L'unica differenza tra queste due opzioni (a parte il nome) è che la query ora deve essere racchiusa tra parentesi e avere un alias, ovvero passare la query originale come sottoquery.
2. numPartitions:
   il numero massimo di partizioni che possono essere utilizzate per il parallelismo nella lettura e scrittura della tabella. Ciò determina anche il numero massimo di connessioni JDBC simultanee. Se il numero di partizioni da scrivere supera questo limite, lo riduciamo a questo limite chiamando coalesce(numPartitions) prima di scrivere. ( fonte ) Ho il mio impostato su 100 nel cluster EMR, ma una buona pratica sarebbe quella di impostare il numero di partizione tra 1 e 4 volte il numero di core. Tuttavia, non è un numero scolpito nella pietra, quindi prova sempre con valori diversi per vedere cosa funziona per te.
3. partitionColumn:
   questo è probabilmente il parametro più importante e quello che farà la differenza più grande nel tempo di esecuzione delle tue query. Poche cose da tenere a mente:
   - la colonna in questione deve essere numerica, data o timestamp
   - per le migliori prestazioni possibili, i valori della colonna devono essere distribuiti il ​​più uniformemente possibile e avere un'elevata cardinalità. Vogliamo evitare colonne distorte di dati: approfondirò questo aspetto di seguito.
   - se hai più colonne con le proprietà di cui sopra, seleziona la colonna indicizzata.
4. lowerbound, upperbound :
   si noti che lowerBound e upperBound sono usati solo per decidere lo stride della partizione, non per filtrare le righe nella tabella. Quindi tutte le righe nella tabella verranno partizionate e restituite. ( fonte ) Di seguito vedrai come Spark genera ogni partizione. Questo è il motivo per cui non vuoi una colonna a cardinalità bassa come colonna di partizione, ma fornirò un esempio di seguito nel caso in cui non abbia senso ora.

Immagina che la colonna che desideri utilizzare per il partizionamento abbia solo valori 0 e 1: un esempio estremo, ma uno scenario comune nell'ingegneria dei dati in cui i valori booleani True e False vengono convertiti in valori numerici.

Poiché hai solo 0 e 1, usi 0 come limite inferiore e 1 come limite superiore. In base a come vengono generate le partizioni - vedere l'immagine sopra - tutte le righe con il valore 0 verranno inserite nella prima partizione (la clausola WHERE della prima partizione è l'unica clausola in cui 0 non viene filtrata) e tutte le righe con il valore 1 verrà spinto all'ultima partizione (la clausola WHERE dell'ultima partizione è l'unica clausola in cui 1 non viene filtrato) .

Quindi, indipendentemente dal numero di partizioni che crei, verranno popolate solo due partizioni e solo due core esecutori faranno tutto il lavoro, gli altri saranno in uno stato inattivo se non ci sono altri lavori pianificati per loro. Per questo motivo, le prestazioni della tua query subiranno un colpo.

Asimmetria dei dati

Immaginiamo di avere un set di dati con 20 milioni di righe e 30 partizioni, i limiti inferiore e superiore sono 2020-01-01 e 2022-12-31. Ai fini di questo articolo, userò una sessione Spark con 30 core esecutori (il che significa che possiamo eseguire fino a 30 attività contemporaneamente). Tieni presente che Spark assegna un'attività per partizione, quindi ogni partizione verrà elaborata da un core esecutore.

Ogni anno di dati sarà diviso in 10 partizioni e considerando che 1 attività sarà assegnata a 1 partizione, ogni attività assegnata agli anni 2020 e 2021 sarà responsabile dell'elaborazione di 100.000 righe, ma ogni attività assegnata all'anno 2022 dovrà elaborare 1,8 milioni di righe. Si tratta di un aumento del 1700% dei dati che devono essere elaborati da un'attività.

Il risultato è che le prime 20 attività finiranno di elaborare la partizione assegnata in pochissimo tempo e successivamente i 20 core esecutori assegnati alle attività completate rimarranno inattivi (se non ci sono altri lavori in cui è possibile utilizzare i core esecutori inattivi) fino all'ultimo 10 attività completano l'elaborazione del 1700% in più di record. Questo è uno spreco di risorse e un aumento dei costi operativi, soprattutto se si usa Spark in un cluster EMR, Glue Job o Databricks.

Come lo aggiustiamo? La soluzione più semplice: utilizzare una colonna diversa per il partizionamento. Ma se è fondamentale utilizzare questa colonna, possiamo utilizzare due query e due dataframe invece di uno. Avrai ancora problemi con i dati distorti a valle (un problema ricorrente nei sistemi computazionali paralleli) ma affrontiamo un problema alla volta.

La prima query caricherà i primi due anni (i 30 executor core saranno divisi tra il 2020 e il 2021), quindi la seconda query caricherà l'ultimo anno (quindi tutti i 30 executor core verranno ora utilizzati per il 2022). Per questo motivo, una determinata attività associata al 2022 sarà ora responsabile dell'elaborazione di 600.000 righe invece di 1,8 milioni di righe. Si tratta di una riduzione del 66% delle righe che devono essere elaborate da un singolo core esecutore rispetto all'implementazione originale e, per di più, ora utilizziamo tutti i core esecutori della nostra sessione Spark invece di lasciarli inattivi.

Cosa succede se non ho alcuna colonna numerica, data o timestamp o la colonna contiene dati distorti?

Non è un problema. In qualità di ingegneri dei dati, utilizzare un CTE e una funzione finestra per generare una colonna numerica row_number()dovrebbe essere un gioco da ragazzi. Verrà introdotto un sovraccarico quando viene generata la nuova colonna, ma anche con questo sovraccarico, la query finale verrà eseguita molto più velocemente utilizzando la colonna appena creata anziché provare a caricare tutti i dati in una partizione utilizzando un core esecutore.

with cte as
(
    SELECT f1, f2, f3
    FROM table
)
select *, row_number() over (order by f1) as rn from cte

original_query = "SELECT f1, f2, f3 FROM table"

final_query = f'(with cte as ({original_query}) select *, row_number() over (order by f1) as rn from cte) as q'

Eseguiamo due query invece di una:
- la prima query verrà eseguita per trovare i valori min e max della colonna che useremo come partitionColumn -
nella seconda query passeremo i risultati della prima query come argomenti lowerbound e upperbound .

query_min_max = f"SELECT min(date), max(date) from ({query}) q"
final_query = f"({query}) as q"

df_min_max = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("query", query_min_max) \
    .load()

min = df_min_max.first()["min"]
max = df_min_max.first()["max"]

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("dbtable", final_query) \
    .option("numPartitions", partitionsCount) \
    .option("partitionColumn", "date") \
    .option("lowerBound", f"{min}") \
    .option("upperBound", f"{max}") \
    .load()

• Se segui il processo di implementazione di Spark nella tua applicazione, prova a usarlo al massimo e leggi i dati contemporaneamente. L'esecuzione di una query SQL su spark.read() come mostrato nella documentazione JDBC di Spark invierà tutti i dati in una partizione e utilizzerà un solo core esecutore, indipendentemente dal numero di core impostati nella sessione Spark.
• Se oggi usi Spark per leggere i dati dalle origini dati JDBC utilizzando l'approccio a thread singolo, prova a eseguire il refactoring delle tue query e migliora le loro prestazioni, potresti persino ottenere una promozione mentre ci sei :)
• Alla fine della giornata, il tempo è denaro. Se sei preoccupato per lo sforzo di riscrivere le tue query, posso parlare per esperienza: alcune delle query Spark che ho refactored utilizzando le tecniche sopra descritte ora vengono eseguite > 90% più velocemente di prima. Potete immaginare quale accoglienza positiva abbia avuto questo miglioramento nella mia squadra.

Gabriele