python - pySpark-在分组数据中查找通用值

我试图在通过应用groupBy创建的组中找到共同的价值观，并在pySpark中的数据框上旋转。
例如，数据如下所示：

+--------+---------+---------+
|PlayerID|PitcherID|ThrowHand|
+--------+---------+---------+
|10000598| 10000104|        R|
|10000908| 10000104|        R|
|10000489| 10000104|        R|
|10000734| 10000104|        R|
|10006568| 10000104|        R|
|10000125| 10000895|        L|
|10000133| 10000895|        L|
|10006354| 10000895|        L|
|10000127| 10000895|        L|
|10000121| 10000895|        L|

申请后：

df.groupBy('PlayerID').pivot('ThrowHand').agg(F.count('ThrowHand')).drop('null').show(10)

我得到类似的东西：

+--------+----+---+
|PlayerID| L  |  R|
+--------+----+---+
|10000591|  11| 43|
|10000172|  22|101|
|10000989|  05| 19|
|10000454|  05| 17|
|10000723|  11| 33|
|10001989|  11| 38|
|10005243|  20| 60|
|10003366|  11| 26|
|10006058|  02| 09|
+--------+----+---+

有什么办法可以让我在上面的L和R的计数中获得'PitcherID'的通用值。

我的意思是，对于PlayerID = 10000591，我有11个PitcherID（其中ThrowHand为L）和43 PitcherID（其中ThrowHand为43）。在这11个和43个投手中，有一些投手是相同的。

有什么办法可以获取这些常见的PitcherID？

最佳答案

您应该首先获取每个掷手的pitcherIds集合，如下所示：

import pyspark.sql.functions as F
#collect set of pitchers in addition to count of ThrowHand
df = df.groupBy('PlayerID').pivot('ThrowHand').agg(F.count('ThrowHand').alias('count'), F.collect_set('PitcherID').alias('PitcherID')).drop('null')

这应该给你dataframe作为

root
 |-- PlayerID: string (nullable = true)
 |-- L_count: long (nullable = false)
 |-- L_PitcherID: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- R_count: long (nullable = false)
 |-- R_PitcherID: array (nullable = true)
 |    |-- element: string (containsNull = true)

然后编写一个udf函数以获取常见的pitcherID作为

#columns with pitcherid and count
pitcherColumns = [x for x in df.columns if 'PitcherID' in x]
countColumns = [x for x in df.columns if 'count' in x]

#udf function to find the common pitcher between the collected pitchers
@F.udf(T.ArrayType(T.StringType()))
def commonFindingUdf(*pitcherCols):
    common = pitcherCols[0]
    for pitcher in pitcherCols[1:]:
        common = set(common).intersection(pitcher)
    return [x for x in common]

#calling the udf function and selecting the required columns
df.select(F.col('PlayerID'), commonFindingUdf(*[col(x) for x in pitcherColumns]).alias('common_PitcherID'), *countColumns)

这应该给你最终的dataframe作为

root
 |-- PlayerID: string (nullable = true)
 |-- common_PitcherID: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- L_count: long (nullable = false)
 |-- R_count: long (nullable = false)

我希望答案是有帮助的