Pyspark DataFrame - Wie man Variablen benutzt, um beizutreten?

Question

Ich habe ein bisschen Mühe, eine Verbindung auf zwei Datenrahmen mit Spark Data Frames auf Python zu machen. Ich habe zwei Datenrahmen, die ich den Namen der Spalten ändern musste, um sie für jeden Datenrahmen einzigartig zu machen, so dass ich später sagen konnte, welche Spalte was ist. Ich tat dies, um die Spalten umbenennen (firstDf und secondDf ist Spark-Datenrahmen mit der Funktion createDataFrame erstellt):

oldColumns = firstDf.schema.names 
newColumns = list(map(lambda x: "{}.{}".format('firstDf', x), oldColumns)) 
firstDf = firstDf.toDF(*newColumns) 

ich dies für den zweiten Datenrahmen wiederholt. Dann habe ich versucht, sie zu verbinden, mit dem folgenden Code:

from pyspark.sql.functions import * 

firstColumn = 'firstDf.firstColumn' 
secondColumn = 'secondDf.firstColumn' 
joinedDF = firstDf.join(secondDf, col(firstColumn) == col(secondColumn), 'inner') 

es wie folgt verwenden ich die folgende Fehlermeldung erhalten:

AnalysisException "cannot resolve 'firstDf.firstColumn' given input columns: [firstDf.firstColumn, ...];"

Dies war nur zu zeigen, dass die Spalte in dem Feld Eingabespalten vorhanden .

Wenn ich nicht umbenennen, die Datenrahmen Spalten Ich bin in der Lage, sie zu verbinden Code mit diesem Stück:

joinedDf = firstDf.join(secondDf, firstDf.firstColumn == secondDf.firstColumn, 'inner') 

Aber geben Sie mir einen Datenrahmen mit mehrdeutigen Spaltennamen.

Irgendwelche Ideen, wie man das angeht?

Answer 1

Im Allgemeinen keine Punkte in Namen verwenden. Diese haben eine besondere Bedeutung (können entweder zum Ermitteln der Tabelle oder zum Zugriff auf struct Felder verwendet werden) und erfordern zusätzliche Arbeit, um korrekt erkannt zu werden.

Für equi verbindet alles, was Sie brauchen, ist ein Spaltenname ist:

from pyspark.sql.functions import col 

firstDf = spark.createDataFrame([(1, "foo")], ("firstColumn", "secondColumn")) 
secondDf = spark.createDataFrame([(1, "foo")], ("firstColumn", "secondColumn")) 

column = 'firstColumn' 
firstDf.join(secondDf, [column], 'inner') 

## DataFrame[firstColumn: bigint, secondColumn: string, secondColumn: string] 

Für komplexe Fälle verwenden Tabellen-Aliases:

firstColumn = 'firstDf.firstColumn' 
secondColumn = 'secondDf.firstColumn' 

firstDf.alias("firstDf").join(
    secondDf.alias("secondDf"), 
    # After alias prefix resolves to table name 
    col(firstColumn) == col(secondColumn), 
    "inner" 
) 

## DataFrame[firstColumn: bigint, secondColumn: string, firstColumn: bigint, secondColumn: string] 

Sie auch übergeordnete Rahmen direkt verwenden:

column = 'firstColumn' 

firstDf.join(secondDf, firstDf[column] == secondDf[column])