Spark Dataframe odróżnić kolumny z powtórzoną nazwą

Question

Więc co wiem w Spark Dataframe, że dla wielu kolumn mogą mieć taką samą nazwę jak pokazano poniżej dataframe migawka:

[ 
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), 
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=125231, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0047, 3: 0.0, 4: 0.0043})), 
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=145831, f=SparseVector(5, {0: 0.0, 1: 0.2356, 2: 0.0036, 3: 0.0, 4: 0.4132})), 
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=147031, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), 
Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=149231, f=SparseVector(5, {0: 0.0, 1: 0.0032, 2: 0.2451, 3: 0.0, 4: 0.0042})) 
] 

Powyższy wynik jest tworzony przez przyłączyć z dataframe do siebie, widać, że są kolumny 4 z dwoma a i f.

Problemem jest to że gdy próbuję wykonać więcej obliczeń z kolumną a, nie mogę znaleźć sposób, aby wybrać a, muszę spróbować df[0] i df.select('a'), zarówno zwrócone mnie poniżej błędu mesaage:

AnalysisException: Reference 'a' is ambiguous, could be: a#1333L, a#1335L. 

Czy mimo to w interfejsie API Sparka mogę ponownie odróżnić kolumny od powtórzonych nazw? a może jakiś sposób na zmianę nazw kolumn?

Answer 1

Polecam, zmienić nazwy kolumn dla join

df1.select('a as "df1_a", 'f as "df1_f") 
    .join(df2.select('a as "df2_a", 'f as "df2_f"), 'df1_a === 'df2_a) 

Powstały DataFrame będzie miał schema

(df1_a, df1_f, df2_a, df2_f) 

Answer 2

Po wykopaniu do API Spark, znalazłem mogę najpierw użyć alias stworzyć alias dla oryginalnego dataframe następnie użyć withColumnRename ręcznie zmienić nazwę każdej kolumny na aliasu w końcu zrobić join bez powodowania nazwę kolumny powielanie.

Więcej szczegółów można zapoznać się poniżej Spark Dataframe API:

pyspark.sql.DataFrame.alias

pyspark.sql.DataFrame.withColumnRenamed

Myślę jednak, że to tylko kłopotliwe obejście, i zastanawiałem się, czy jest jakiś lepszy sposób na moje pytanie .

Answer 3

Zacznijmy od jakiegoś danych:

from pyspark.mllib.linalg import SparseVector 
from pyspark.sql import Row 

df1 = sqlContext.createDataFrame([ 
    Row(a=107831, f=SparseVector(
     5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), 
    Row(a=125231, f=SparseVector(
     5, {0: 0.0, 1: 0.0, 2: 0.0047, 3: 0.0, 4: 0.0043})), 
]) 

df2 = sqlContext.createDataFrame([ 
    Row(a=107831, f=SparseVector(
     5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), 
    Row(a=107831, f=SparseVector(
     5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), 
]) 

Istnieje kilka sposobów s możesz podejść do tego problemu.Przede wszystkim można jednoznacznie odwołać podrzędne kolumny tabeli przy użyciu kolumny nadrzędny:

df1.join(df2, df1['a'] == df2['a']).select(df1['f']).show(2) 

## +--------------------+ 
## |     f| 
## +--------------------+ 
## |(5,[0,1,2,3,4],[0...| 
## |(5,[0,1,2,3,4],[0...| 
## +--------------------+ 

Można również używać aliasów tabel:

from pyspark.sql.functions import col 

df1_a = df1.alias("df1_a") 
df2_a = df2.alias("df2_a") 

df1_a.join(df2_a, col('df1_a.a') == col('df2_a.a')).select('df1_a.f').show(2) 

## +--------------------+ 
## |     f| 
## +--------------------+ 
## |(5,[0,1,2,3,4],[0...| 
## |(5,[0,1,2,3,4],[0...| 
## +--------------------+ 

Wreszcie można programowo zmienić nazwę kolumny:

df1_r = df1.select(*(col(x).alias(x + '_df1') for x in df1.columns)) 
df2_r = df1.select(*(col(x).alias(x + '_df2') for x in df2.columns)) 

df1_r.join(df2_r, col('a_df1') == col('a_df2')).select(col('f_df1')).show(2) 

## +--------------------+ 
## |    f_df1| 
## +--------------------+ 
## |(5,[0,1,2,3,4],[0...| 
## |(5,[0,1,2,3,4],[0...| 
## +--------------------+ 

Answer 4

You można użyć metody def drop(col: Column), aby usunąć zduplikowaną kolumnę, na przykład:

DataFrame:df1 

+-------+-----+ 
| a  | f | 
+-------+-----+ 
|107831 | ... | 
|107831 | ... | 
+-------+-----+ 

DataFrame:df2 

+-------+-----+ 
| a  | f | 
+-------+-----+ 
|107831 | ... | 
|107831 | ... | 
+-------+-----+ 

kiedy łączę DF1 z df2 The DataFrame będzie jak poniżej:

val newDf = df1.join(df2,df1("a")===df2("a")) 

DataFrame:newDf 

+-------+-----+-------+-----+ 
| a  | f | a  | f | 
+-------+-----+-------+-----+ 
|107831 | ... |107831 | ... | 
|107831 | ... |107831 | ... | 
+-------+-----+-------+-----+ 

Teraz możemy użyć def drop(col: Column) metodę do spadku kopiowane kolumny 'A' lub 'F', podobnie jak w następujący sposób:

val newDfWithoutDuplicate = df1.join(df2,df1("a")===df2("a")).drop(df2("a")).drop(df2("f"))