在Spark中加入DF后删除重复列

Question

当您连接具有相似列名称的两个DF时:

df = df1.join(df2, df1['id'] == df2['id'])

加入工作正常,但你不能调用id列,因为它是不明确的,你会得到以下异常:

pyspark.sql.utils.AnalysisException: "Reference 'id' is ambiguous, could be: id#5691, id#5918.;"

这使得id不再可用......

以下函数解决了该问题:

def join(df1, df2, cond, how='left'):
    df = df1.join(df2, cond, how=how)
    repeated_columns = [c for c in df1.columns if c in df2.columns]
    for col in repeated_columns:
        df = df.drop(df2[col])
    return df

我不喜欢它的是我必须迭代列名称并删除它们为什么由一个.这看起来很笨重......

您是否知道任何其他解决方案将更优雅地加入和删除重复项或删除多个列而不迭代它们？

Answer 1

如果两个数据框的连接列具有相同的名称,并且您只需要equi连接,则可以将连接列指定为列表,在这种情况下,结果将只保留其中一个连接列:

df1.show()
+---+----+
| id|val1|
+---+----+
|  1|   2|
|  2|   3|
|  4|   4|
|  5|   5|
+---+----+

df2.show()
+---+----+
| id|val2|
+---+----+
|  1|   2|
|  1|   3|
|  2|   4|
|  3|   5|
+---+----+

df1.join(df2, ['id']).show()
+---+----+----+
| id|val1|val2|
+---+----+----+
|  1|   2|   2|
|  1|   2|   3|
|  2|   3|   4|
+---+----+----+

否则,您需要提供连接数据帧别名,并在以后通过别名引用重复的列:

df1.alias("a").join(
    df2.alias("b"), df1['id'] == df2['id']
).select("a.id", "a.val1", "b.val2").show()
+---+----+----+
| id|val1|val2|
+---+----+----+
|  1|   2|   2|
|  1|   2|   3|
|  2|   3|   4|
+---+----+----+

Answer 2

假设“a”是一个带有“id”列的数据框，“b”是另一个带有“id”列的数据框

我使用以下两种方法来删除重复项：

方法 1：使用字符串连接表达式而不是布尔表达式。这会自动为您删除重复的列

a.join(b, 'id')

方法 2：在加入之前重命名列并在加入之后删除它

b.withColumnRenamed('id', 'b_id')
joinexpr = a['id'] == b['b_id']
a.join(b, joinexpr).drop('b_id)

Answer 3

下面的代码适用于 Spark 1.6.0 及更高版本。

salespeople_df.show()
+---+------+-----+
|Num|  Name|Store|
+---+------+-----+
|  1| Henry|  100|
|  2| Karen|  100|
|  3|  Paul|  101|
|  4| Jimmy|  102|
|  5|Janice|  103|
+---+------+-----+

storeaddress_df.show()
+-----+--------------------+
|Store|             Address|
+-----+--------------------+
|  100|    64 E Illinos Ave|
|  101|         74 Grand Pl|
|  102|          2298 Hwy 7|
|  103|No address available|
+-----+--------------------+

假设（在此示例中）共享列的名称相同：

joined=salespeople_df.join(storeaddress_df, ['Store'])
joined.orderBy('Num', ascending=True).show()

+-----+---+------+--------------------+
|Store|Num|  Name|             Address|
+-----+---+------+--------------------+
|  100|  1| Henry|    64 E Illinos Ave|
|  100|  2| Karen|    64 E Illinos Ave|
|  101|  3|  Paul|         74 Grand Pl|
|  102|  4| Jimmy|          2298 Hwy 7|
|  103|  5|Janice|No address available|
+-----+---+------+--------------------+

.join将防止共享列的重复。

假设您想删除Num本例中的列，您可以使用.drop('colname')

joined=joined.drop('Num')
joined.show()

+-----+------+--------------------+
|Store|  Name|             Address|
+-----+------+--------------------+
|  103|Janice|No address available|
|  100| Henry|    64 E Illinos Ave|
|  100| Karen|    64 E Illinos Ave|
|  101|  Paul|         74 Grand Pl|
|  102| Jimmy|          2298 Hwy 7|
+-----+------+--------------------+

Answer 4

df.join(other, on, how)当on是列名字符串或列名字符串列表时，返回的数据框将防止重复的列。当on是连接表达式时，将导致重复的列。我们可以使用.drop(df.a)删除重复的列。例：

cond = [df.a == other.a, df.b == other.bb, df.c == other.ccc]
# result will have duplicate column a
result = df.join(other, cond, 'inner').drop(df.a)