Pyspark Multiple JOINS Column <> Row values: Reducing Actions

Question

Ho una tabella principale "Tabella 1" con 3 colonne (mostrata sotto). Le tabelle 2.1, 3.1 e 4.1 si riferiscono a 3 date uniche presenti nella Tabella 1 e devono essere compilate nella colonna "Punti 1". Allo stesso modo, le tabelle 2.2, 3.2 e 4.2 si riferiscono alle stesse 3 date uniche presenti nella tabella 1 e devono essere inserite nella colonna "Punti 2".

Approccio attuale:

df1 = spark.table("Table1")
df2_1 = spark.table("table2.1")
df2_1 = withColumn("Date", lit(3312019))


df3 = df1.join(df2_1, df1.ID==df2.1==ID & df1.Date==df2_1.Date, 'left')
df4 = df3.withColumn('Points', when(df3.Category==A, col('A'))
                              .when(df3.Category==B, col('B'))
                              .when(df3.Category==C, col('C'))
                              .when(df3.Category==D, col('D'))
                              .otherwise(lit(None)))

L'approccio attuale rende il mio codice lungo se implementato per tutte e 6 le tabelle, qualche suggerimento per accorciarlo e ridurre più azioni?

Answer 1

Non so se sia molto più breve o "più pulito" della tua versione, ma dato che hai chiesto aiuto su questo, posterò questa come risposta. Si prega di notare che la mia risposta è in normale spark (scala) - non pyspark, ma non dovrebbe essere troppo difficile portarlo su pyspark, se trovi utile la risposta :)

Quindi ecco qui:

Prima una piccola funzione di aiuto

def columns2rows(row: Row) = {
  val id = row.getInt(0)
  val date = row.getInt(1)
  val cols = Seq("A", "B", "C", "D")
  cols.indices.map(index => (id, cols(index), date, if (row.isNullAt(index+2)) 0 else row.getInt(index+2)))
}

Quindi unisci le tabelle necessarie per popolare "Punti1"

val df1 = table21.withColumn("Date", lit(3312019))
  .unionByName(table31.withColumn("Date", lit(12312019)))
  .unionByName(table41.withColumn("Date", lit(5302020)))
  .select($"ID", $"Date", $"A", $"B", $"C", $"D")
  .flatMap(row => columns2rows(row))
  .toDF("ID", "Category", "Date", "Points1")

Quindi unisci le tabelle necessarie per popolare "Points2"

val df2 = table22.withColumn("Date", lit(3312019))
  .unionByName(table32.withColumn("Date", lit(12312019)))
  .unionByName(table42.withColumn("Date", lit(5302020)))
  .select($"ID", $"Date", $"A", $"B", $"C", $"D")
  .flatMap(row => columns2rows(row))
  .toDF("ID", "Category", "Date", "Points2")

Uniscili insieme e finalmente con il tavolo originale:

val joiningTable = df1.join(df2, Seq("ID", "Category", "Date"))

val res = table1.join(joiningTable, Seq("ID", "Category", "Date"))

... e voilà - stampa il risultato finale:

res.show()

+---+--------+--------+-------+-------+
| ID|Category|    Date|Points1|Points2|
+---+--------+--------+-------+-------+
|123|       A| 3312019|     40|     20|
|123|       B| 5302020|     10|     90|
|123|       D| 5302020|      0|     80|
|123|       A|12312019|     20|     10|
|123|       B|12312019|      0|     10|
|123|       B| 3312019|     60|     60|
+---+--------+--------+-------+-------+