为 pyspark 数据帧的每一行评估多个 if elif 条件答案

【问题标题】：Multiple if elif conditions to be evaluated for each row of pyspark dataframe为 pyspark 数据帧的每一行评估多个 if elif 条件
【发布时间】：2017-07-25 05:42:57
【问题描述】：

我需要 pyspark 数据框主题方面的帮助。我有一个说 1000+ 列和 100000+ 行的数据框。此外，如果存在 elif 条件，我还有 10000+，在每个 if else 条件下，很少有全局变量会增加一些值。现在我的问题是如何仅在 pyspark 中实现这一点。我阅读了有关过滤器以及根据条件返回行的函数的内容，我需要检查那些 10000+ if else 条件并执行一些操作。

任何帮助将不胜感激。

如果你能举一个小数据集的例子，那会很有帮助。

谢谢

【问题讨论】：

请提供您已经尝试过的内容的概述，以及您希望在某些玩具问题上实现的条件。

标签： apache-spark pyspark spark-dataframe bigdata

【解决方案1】：

您可以定义一个函数来包含所有 if elif 条件，然后将此函数应用于 DataFrame 的每一行。

只需使用.rdd 将DataFrame 转换为普通RDD，然后使用map() 函数即可。
例如，DF.rdd.map(lambda row: func(row))

希望对你有帮助。

【讨论】：

【解决方案2】：

据我了解，您只想在迭代 DataFrame 时更新一些全局计数器。为此，您需要：

1) 定义一个或多个accumulators:

ac_0 = sc.accumulator(0)
ac_1 = sc.accumulator(0)

2) 定义一个函数来更新给定行的累加器，例如：

def accumulate(row):
    if row.foo:
        ac_0.add(1)
    elif row.bar:
        ac_1.add(row.baz)

3) 在您的 DataFrame 上调用 `foreach`：

df.foreach(accumulate)

4) 检查累加器值

> ac_0.value
>>> 123

【讨论】：

1) 定义一个或多个accumulators:

2) 定义一个函数来更新给定行的累加器，例如：

3) 在您的 DataFrame 上调用 foreach：

4) 检查累加器值

3) 在您的 DataFrame 上调用 `foreach`：