Spark Dataframe区分具有重复名称的列

Question

正如我在Spark 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}))
]

上面的结果是通过将数据框连接到自身来创建的,你可以看到有4两个a和两个列f.

问题是,当我尝试使用a列进行更多计算时,我无法找到一种方法来选择a,我已经尝试了,df[0]并且df.select('a')都返回了我的错误评论:

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

无论如何在Spark API中我可以再次将列与重复的名称区分开来吗？或者某种方式让我改变列名？

Answer 1

让我们从一些数据开始:

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})),
])

有几种方法可以解决这个问题.首先,您可以使用父列明确引用子表列:

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...|
##  +--------------------+

您还可以使用表别名:

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...|
##  +--------------------+

最后,您可以以编程方式重命名列:

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...|
## +--------------------+

Answer 2

我建议你更改你的列名 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)

结果DataFrame将有schema

(df1_a, df1_f, df2_a, df2_f)

Answer 3

有一种比为正在加入的所有列编写别名更简单的方法:

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

如果您加入的密钥在两个表中都相同,则此方法有效.

请参阅 https://docs.databricks.com/spark/latest/faq/join-two-dataframes-duplicated-column.html

Answer 4

假设您要连接的 DataFrame 是 df1 和 df2，并且您将它们连接到列“a”上，那么您有 2 个方法

方法一

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

这是一个很棒的方法，强烈推荐。

方法2

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

Answer 5

这就是我们如何在PySpark 中以相同的列名连接两个Dataframe。

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

如果您printSchema()在此之后这样做，那么您可以看到重复的列已被删除。

Answer 6

在深入了解Spark API之后,我发现我可以首先使用alias为原始数据帧创建别名,然后我使用withColumnRenamed手动重命名别名上的每一列,这样做会join导致列名重复.

更多细节可以参考Spark Dataframe API:

pyspark.sql.DataFrame.alias

pyspark.sql.DataFrame.withColumnRenamed

但是,我认为这只是一个麻烦的解决方法,并想知道我的问题是否有更好的方法.

Answer 7

您可以使用def drop(col: Column)方法删除重复的列，例如：

DataFrame:df1

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

DataFrame:df2

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

当我将df1与df2结合使用时，DataFrame将如下所示：

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

DataFrame:newDf

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

现在，我们可以使用def drop(col: Column)method删除重复的列'a'或'f'，如下所示：

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

Answer 8

如果两个表中只有键列相同，则尝试使用以下方法（方法 1）：

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

而不是下面（方法 2）：

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

使用方法 1 的优点：

• “key”将在最终数据框中仅显示一次
• 易于使用的语法

使用方法 1 的缺点：

• 只对关键列有帮助
• 场景，其中左连接的情况下，如果打算使用右键空计数，这是行不通的。在这种情况下，必须如上所述重命名其中一个键。