将 PySpark Dataframe 过滤器的结果收集到一个变量中答案

【问题标题】：Collecting the result of PySpark Dataframe filter into a variable将 PySpark Dataframe 过滤器的结果收集到一个变量中
【发布时间】：2018-08-31 21:45:41
【问题描述】：

我正在使用PySpark 数据框。我的数据集包含三个属性，id、name 和 address。我正在尝试根据 name 值删除相应的行。我一直在尝试的是获得我要删除的行的唯一id

ID = df.filter(df["name"] == "Bruce").select(df["id"]).collect()

我得到的输出如下：[Row(id='382')]

我想知道如何使用id 删除一行。另外，我怎样才能用另一个替换数据框中的某些值？例如，将所有values == "Bruce" 替换为"John"

【问题讨论】：

您只需要从输出中提取 ID 值。 id_vals = [r['id'] for r in ID]
您不能在 DataFrame 中使用索引。 Spark DataFrames 具有与 pandas 不同的结构行为。 Spark 也不是 RDBMS。所以除非你过滤和收集，否则没什么可做的。 Spark DataFrame 不适合用作查找表。
您可以使用 head() 和 map 将收集到的条目转换为一个条目。
谢谢@pault。这就是我想知道的。
@zimmer np，我添加了一个带有解释的答案。我忍不住想这是XY Problem。你想做什么？可能有比使用 collect 一次收集一个 id 更好的解决方案。

标签： python dataframe pyspark

【解决方案1】：

来自pyspark.sql.DataFrame.collect() 的文档，函数：

将所有记录作为 Row 列表返回。

pyspark.sql.Row 中的字段可以像字典值一样被访问。

所以对于你的例子：

ID = df.filter(df["name"] == "Bruce").select(df["id"]).collect()
#[Row(id='382')]

您可以通过以下方式访问id 字段：

id_vals = [r['id'] for r in ID]
#['382']

但是一次查找一个值对于 spark DataFrames 来说通常是一个不好的用途。你应该考虑一下你的最终目标，看看是否有更好的方法来实现它。

编辑

根据您的 cmets，您似乎想将 name 列中的值替换为另一个值。一种方法是使用pyspark.sql.functions.when()。

此函数将布尔列表达式作为第一个参数。我正在使用f.col("name") == "Bruce"。如果布尔表达式为True，则第二个参数应返回。对于这个例子，我使用的是f.lit(replacement_value)。

例如：

import pyspark.sql.functions as f
replacement_value = "Wayne"
df = df.withColumn(
    "name",
    f.when(f.col("name") == "Bruce", f.lit(replacement_value)).otherwise(f.col("name"))
)

【讨论】：

谢谢@pault。我试图删除名为“Bruce”的行。我知道我的方法太复杂了，但我想出了一个更简单的方法。这里：df = df[df["name"] != "Bruce"] 我也可以用另一个字符串替换特定名称吗？
当我编写上面的代码时，我收到了TypeError: condition should be a Column。我也试过==。有什么我可能做错了吗？
@zimmer 我使用pyspark.sql.functions.col 使用更明确的语法为答案添加了一个编辑。