Spark Dataframe rozróżnia kolumny ze zduplikowaną nazwą

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

[
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 połączenie z ramką danych do samego siebie, możesz zobaczyć, że istnieją 4kolumny z dwoma ai f.

Problem polega na tym, że gdy próbuję wykonać więcej obliczeń z akolumną, nie mogę znaleźć sposobu, aby wybrać a, próbowałem df[0]i df.select('a')oba zwróciły mi poniżej komunikat o błędzie:

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

Czy mimo to w interfejsie Spark API mogę ponownie odróżnić kolumny od zduplikowanych nazw? a może jakiś sposób, aby zmienić nazwy kolumn?

Zalecałbym zmianę nazw kolumn w pliku join.

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

Wynikowy DataFramebędzie miałschema

(df1_a, df1_f, df2_a, df2_f)

Zacznijmy od pewnych 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 rozwiązania tego problemu. Przede wszystkim możesz jednoznacznie odwołać się do podrzędnych kolumn tabeli za pomocą kolumn nadrzędnych:

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żesz także użyć 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żesz programowo zmienić nazwy kolumn:

df1_r = df1.select(*(col(x).alias(x + '_df1') for x in df1.columns))
df2_r = df2.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...|
## +--------------------+

Istnieje prostszy sposób niż pisanie aliasów dla wszystkich kolumn, do których się przyłączasz, wykonując:

df1.join(df2,['a'])

Działa to, jeśli klucz, do którego się przyłączasz, jest taki sam w obu tabelach.

Zobacz https://kb.databricks.com/data/join-two-dataframes-duplicated-columns.html

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

DataFrame:df1

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

DataFrame:df2

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

kiedy dołączę df1 do df2, DataFrame będzie wyglądać 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)metody, aby usunąć zduplikowaną kolumnę „a” lub „f”, tak jak poniżej:

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

Po zagłębieniu się w Spark API odkryłem, że mogę najpierw użyć aliasdo utworzenia aliasu dla oryginalnej ramki danych, a następnie używam withColumnRenameddo ręcznej zmiany nazwy każdej kolumny na aliasie, zrobi to joinbez powodowania duplikacji nazwy kolumny.

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

pyspark.sql.DataFrame.alias

pyspark.sql.DataFrame.withColumnRenamed

Uważam jednak, że jest to tylko kłopotliwe obejście i zastanawiam się, czy istnieje lepszy sposób na moje pytanie.

Załóżmy, że ramki danych, które chcesz dołączyć, to df1 i df2, a łączysz je w kolumnie „a”, a następnie masz 2 metody

Metoda 1

df1.join (df2, 'a', 'left_outer')

To świetna metoda i jest wysoce zalecana.

Metoda 2

df1.join (df2, df1.a == df2.a, 'left_outer'). drop (df2.a)

W ten sposób możemy połączyć dwie ramki Dataframe na tych samych nazwach kolumn w PySpark.

df = df1.join(df2, ['col1','col2','col3'])

Jeśli to zrobisz printSchema(), zobaczysz, że zduplikowane kolumny zostały usunięte.

To może nie być najlepsze podejście, ale jeśli chcesz zmienić nazwę zduplikowanych kolumn (po połączeniu), możesz to zrobić za pomocą tej niewielkiej funkcji.

def rename_duplicate_columns(dataframe):
    columns = dataframe.columns
    duplicate_column_indices = list(set([columns.index(col) for col in columns if columns.count(col) == 2]))
    for index in duplicate_column_indices:
        columns[index] = columns[index]+'2'
    dataframe = dataframe.toDF(*columns)
    return dataframe

jeśli tylko kolumna klucza jest taka sama w obu tabelach, spróbuj zastosować następującą metodę (podejście 1):

left. join(right , 'key', 'inner')

zamiast poniżej (podejście 2):

left. join(right , left.key == right.key, 'inner')

Zalety stosowania podejścia 1:

• „klucz” pojawi się tylko raz w końcowej ramce danych
• łatwa w użyciu składnia

Wady stosowania podejścia 1:

• pomoc tylko z kolumną kluczową
• Scenariusze, w których przypadek lewostronnego złączenia, jeśli planujesz użyć prawego klawisza zliczającego wartość null, nie zadziała. W takim przypadku należy zmienić nazwę jednego z kluczy, jak wspomniano powyżej.

Jeśli masz bardziej skomplikowany przypadek użycia niż opisany w odpowiedzi Glennie Helles Sindholt, np. Masz inne / kilka niełączonych nazw kolumn, które są również takie same i chcesz je rozróżnić przy wyborze, najlepiej użyć aliasów, np .:

df3 = df1.select("a", "b").alias("left")\
   .join(df2.select("a", "b").alias("right"), ["a"])\
   .select("left.a", "left.b", "right.b")

df3.columns
['a', 'b', 'b']