Mappa differenza prestazioni () vs withColumn ()

Question

Ho una tabella con oltre 100 colonne. Devo rimuovere le virgolette doppie da alcune colonne. Ho trovato 2 modi per farlo, usando withColumn () e map ()

Utilizzando withColumn ()

cols_to_fix = ["col1", ..., "col20"]
for col in cols_to_fix:
    df = df.withColumn(col, regexp_replace(df[col], "\"", ""))

Utilizzando map ()

def remove_quotes(row: Row) -> Row:
    row_as_dict = row.asDict()
    cols_to_fix = ["col1", ..., "col20"]
    for column in cols_to_fix:
        if row_as_dict[column]:
            row_as_dict[column] = re.sub("\"", "", str(row_as_dict[column]))
    return Row(**row_as_dict)
 
df = df.rdd.map(remove_quotes).toDF(df.schema)

Ecco la mia domanda. Ho scoperto che l'uso di map () richiede circa 4 volte più tempo rispetto a withColumn () su una tabella che ha ~ 25 milioni di record. Apprezzerò davvero se qualsiasi altro utente di stack overflow può spiegare il motivo della differenza di prestazioni, in modo da poter evitare insidie ​​simili in futuro.

Answer 1

in primo luogo, un consiglio: non convertire DataFrame in RDD e basta eseguire df.map (la tua funzione qui), questo potrebbe far risparmiare molto tempo. la pagina seguentehttps://dzone.com/articles/apache-spark-3-reasons-why-you-should-not-use-rdds ci farebbe risparmiare un sacco di tempo, la sua conclusione principale è che RDD è notevolmente lento di DataFrame / Dataset, per non parlare del tempo utilizzato per la conversione da DataFrame a RDD.

parliamo ora di map e withColumn senza alcuna conversione tra DataFrame e RDD. Prima la conclusione: la mappa è solitamente 5 volte più lenta di withColumn. il motivo è che l'operazione di mappa implica sempre la deserializzazione e la serializzazione mentre withColumn può operare sulla colonna di interesse. per essere precisi, l'operazione sulla mappa dovrebbe deserializzare la riga in più parti su cui verrà eseguita l'operazione,

un esempio qui: supponiamo di avere un DataFrame che assomiglia a + -------- + ----------- + | language | users_count | + -------- + ----------- + | Java | 20000 | | Python | 100000 | | Scala | 3000 | + -------- + ----------- + quindi vogliamo incrementare tutti i valori nella colonna users_count di 1, possiamo farlo in questo modo

    df.map(row => {
        val usersCount = row.getInt(1) + 1
        (row.getString(0), usersCount)
}).toDF("language", "users_count_incremented_by_1")

nel codice sopra, dobbiamo prima deserializzare ogni riga per estrarre i valori nella seconda colonna, dopodiché restituiamo i valori modificati e salvarli come DataFrame (questo passaggio richiede la serializzazione di (a, b) in Row (a, b) poiché DataFrame non è altro che un DataSet di righe). per una spiegazione più dettagliata, controlla il seguente eccellente articolohttps://medium.com/@fqaiser94/udfs-vs-map-vs-custom-spark-native-functions-91ab2c154b44

map non può operare sulla colonna stessa ma deve operare sui valori della colonna, per ottenere i valori occorre deserializzare, salvarlo come dataframe necessita di serializzazione.

ma map è ancora di grande utilità: con l'aiuto del metodo map le persone potrebbero implementare operazioni molto sofisticate mentre si potrebbero fare solo operazioni incorporate se usassimo solo withColumn.

Per riassumere, la mappa è più lenta ma più flessibile, withColumn è sicuramente la più efficiente mentre la sua funzionalità è limitata.