PySpark

Question

抱歉，标题含糊不清，我想不出更好的表达方式。我了解一点 python 并且对 Pandas 数据帧有一些经验，但最近我的任务是研究涉及 Spark 的东西，我正在努力取得进展。

我想最好的解释方法是举一个小例子。假设我有数据框 A:

id | Name    |
--------------
1  | Random  |
2  | Random  |
3  | Random  |

以及数据框 B:

id | Fruit  |
-------------
1  | Pear   |
2  | Pear   |
2  | Apple  |
2  | Banana |
3  | Pear   |
3  | Banana |

现在我要做的是将数据框 A 与 B 匹配（基于 ID 匹配），然后遍历数据框 B 中的 Fruit 列。如果出现一个值（比如 Banana），我想添加它作为数据框的一列。可以是一个简单的求和（每次出现 banana 时将 1 添加到列中），或者如果它出现一次则只是 class。因此，例如，输出可能如下所示：

id | Name    | Banana
---------------------
1  | Random  | 0
2  | Random  | 1
3  | Random  | 1

我的问题是遍历 Spark 数据帧，以及如果确实发生匹配，我如何连接两者。我试图为此做些事情：

def fruit(input):

    fruits = {"Banana" : "B"}

    return fruits[input]

fruits = df.withColumn("Output", fruit("Fruit"))

但这并没有真正起作用。有任何想法吗？提前道歉我对Spark的经验很少。

Answer 1

希望对您有所帮助！

#sample data
A = sc.parallelize([(1,"Random"), (2,"Random"), (3,"Random")]).toDF(["id", "Name"])
B = sc.parallelize([(1,"Pear"), (2,"Pear"), (2,"Apple"), (2,"Banana"), (3,"Pear"), (3,"Banana")]).toDF(["id", "Fruit"])

df_temp = A.join(B, A.id==B.id, 'inner').drop(B.id)
df = df_temp.groupby(df_temp.id, df_temp.Name).\
    pivot("Fruit").\
    count().\
    na.fill(0)
df.show()

输出是

+---+------+-----+------+----+
| id|  Name|Apple|Banana|Pear|
+---+------+-----+------+----+
|  1|Random|    0|     0|   1|
|  3|Random|    0|     1|   1|
|  2|Random|    1|     1|   1|
+---+------+-----+------+----+

编辑备注：如果你只对几种水果感兴趣，那么

from pyspark.sql.functions import col
#list of fruits you are interested in
fruit_list = ["Pear", "Banana"]
df = df_temp.\
    filter(col('fruit').isin(fruit_list)).\
    groupby(df_temp.id, df_temp.Name).\
    pivot("Fruit").\
    count().\
    na.fill(0)
df.show()


+---+------+------+----+
| id|  Name|Banana|Pear|
+---+------+------+----+
|  1|Random|     0|   1|
|  3|Random|     1|   1|
|  2|Random|     1|   1|
+---+------+------+----+

PySpark - 如果值出现在两个字段匹配的另一个数据帧上，则将值返回到数据帧中

PySpark - Returning a value into a dataframe, if a value occurs on another dataframe where two fields match

apache-spark

spark-dataframe