在PySpark中對具有陣列的列進行分組和聚合

我有以下 PySpark 資料框。column_2 是復雜資料型別 array<map<string,bigint>>

Column_1 Column_2                 Column_3
A        [{Mat=7},{Phy=8}]        ABC
A        [{Mat=7},{Phy=8}]        CDE
B        [{Mat=6},{Phy=7}]        ZZZ

我必須對第 1 列和第 2 列進行分組，并獲得第 3 列的最小聚合。

問題是當我嘗試按第 1 列和第 2 列分組時，它給了我一個錯誤

不能用作分組運算式，因為資料型別不是可排序的資料型別

有沒有辦法將此列包含在分組中或以某種方式聚合它。column_2 中的值對于 column_1 中的鍵值將始終相同

預期輸出：

Column_1 Column_2                Column_3
A        [{Mat=7},{Phy=8}]       ABC
B        [{Mat=6},{Phy=7}]       ZZZ

是否可以在聚合函式中收集所有值的串列并將其展平并洗掉重復項？

uj5u.com熱心網友回復：

column_2 中的值對于 column_1 中的鍵值將始終相同

如果是這樣，那么您可以只取first組中的值。

測驗資料框：

from pyspark.sql import functions as F

df = spark.createDataFrame(
    [('A', 'ABC', 7, 8),
     ('A', 'CDE', 7, 8),
     ('B', 'ZZZ', 6, 7)],
    ['Column_1', 'Column_3', 'm', 'p'])
df = df.select(
    'Column_1',
    F.array(F.create_map(F.lit('Mat'), 'm'), F.create_map(F.lit('Phy'), 'p')).alias('Column_2'),
    'Column_3'
)
df.show(truncate=False)
print(df.dtypes)
#  -------- ------------------------ -------- 
# |Column_1|Column_2                |Column_3|
#  -------- ------------------------ -------- 
# |A       |[{Mat -> 7}, {Phy -> 8}]|ABC     |
# |A       |[{Mat -> 7}, {Phy -> 8}]|CDE     |
# |B       |[{Mat -> 6}, {Phy -> 7}]|ZZZ     |
#  -------- ------------------------ -------- 

# [('Column_1', 'string'), ('Column_2', 'array<map<string,bigint>>'), ('Column_3', 'string')]

聚合：

df2 = df.groupBy('Column_1').agg(
    F.first('Column_2').alias('Column_2'),
    F.min('Column_3').alias('Column_3')
)
df2.show(truncate=False)
#  -------- ------------------------ -------- 
# |Column_1|Column_2                |Column_3|
#  -------- ------------------------ -------- 
# |A       |[{Mat -> 7}, {Phy -> 8}]|ABC     |
# |B       |[{Mat -> 6}, {Phy -> 7}]|ZZZ     |
#  -------- ------------------------ -------- 

uj5u.com熱心網友回復：

我可能誤解了你的問題。如果我這樣做了，任何人都不會受到傷害。

我以為您想選擇Column_2. 因此，我稍微修改了資料框以確保組A具有多個值。見 df

df = spark.createDataFrame(
    [('A', 'ABC', 7, 8),
     ('A', 'CDE', 3, 8),
     ('B', 'ZZZ', 6, 7)],
    ['Column_1', 'Column_3', 'm', 'p'])
df = df.select(
    'Column_1',
    F.array(F.create_map(F.lit('Mat'), 'm'), F.create_map(F.lit('Phy'), 'p')).alias('Column_2'),
    'Column_3'
)
df.show(truncate=False)  

df

 -------- ------------------------ -------- 
|Column_1|Column_2                |Column_3|
 -------- ------------------------ -------- 
|A       |[{Mat -> 7}, {Phy -> 8}]|ABC     |
|A       |[{Mat -> 3}, {Phy -> 8}]|CDE     |
|B       |[{Mat -> 6}, {Phy -> 7}]|ZZZ     |
 -------- ------------------------ -------- 

解決方案

如果我的假設是正確的

1. 將鍵值對的值提取Column_2到名為filter
2. 通過對它們求和來聚合它們。存出進來filter
3. 訂購方式Column_1和filter
4. 洗掉帶有子集的重復項Column_1

下面的代碼

new = df.withColumn("filter",F.expr("aggregate(transform(Column_2,x -> map_values(x)[0] ),cast(0 as bigint),(x,i)->x i)")).orderBy('Column_1',desc('filter')).dropDuplicates(['Column_1']).drop('filter')
new.show()
    
 -------- ------------------------ -------- 
|Column_1|Column_2                |Column_3|
 -------- ------------------------ -------- 
|A       |[{Mat -> 7}, {Phy -> 8}]|ABC     |
|B       |[{Mat -> 6}, {Phy -> 7}]|ZZZ     |
 -------- ------------------------ -------- 

標籤：数组 阿帕奇火花 pyspark apache-spark-sql 聚合函数

上一篇：PySpark，如何處理并非總是創建以避免AnalysisException的列？

下一篇：AccessControlException：使用Hive倉庫時，客戶端無法通過：[TOKEN，KERBEROS]進行身份驗證