python之PySpark:我们应该迭代更新数据框吗_编程设计

我的问题有两个部分。第一个是了解 Spark 的工作方式，第二个是优化。

我有一个包含多个分类变量的 spark 数据框。对于这些分类变量中的每一个，我都添加了一个新列，其中每一行都是相应级别的频率。

例如

Date_Built  Square_Footage  Num_Beds    Num_Baths   State   Price     Freq_State 
01/01/1920  1700            3           2           NY      700000    4500

在这里，对于 State(分类变量)，我添加了一个新变量 Freq_State。 NY 级别在数据集中出现 4500 次，因此该行在 Freq_State 列中得到 4500。

我有多个这样的列，我在其中添加了对应级别的列轴承频率。

这是我用来实现这个的代码

def calculate_freq(df, categorical_cols): 
    for each_cat_col in categorical_cols: 
        _freq = df.select(each_cat_col).groupBy(each_cat_col).count() 
        df = df.join(_freq, each_cat_col, "inner") 
    return df

第 1 部分

在这里，如您所见，我正在 for 循环中更新数据框。当我在集群上运行这段代码时，这种更新数据帧的方式是否可取？如果它是一个 Pandas 数据框，我就不会担心这个。但我不确定上下文何时会发生变化。

此外，如果我只是在循环中而不是在函数中运行上述过程，会有什么不同吗？

第 2 部分

有没有更优化的方法来做到这一点？每次我进入循环时我都加入这里？这可以避免吗

请您参考如下方法:

Is there a more optimized way to do this?

有哪些可能的替代方案？

你可以使用窗口函数

def calculate_freq(df, categorical_cols): 
    for cat_col in categorical_cols: 
        w = Window.partitionBy(cat_col) 
        df = df.withColumn("{}_freq".format(each_cat_col), count("*").over(w)) 
    return df

你应该吗？不。与 join 不同，它始终需要对非聚合的 DataFrame 进行完全洗牌。

你可以 melt并使用单个本地对象(这要求所有分类列都属于同一类型):

from itertools import groupby 
 
for c in categorical_cols: 
     df = df.withColumn(c, df[c].cast("string")) 
 
 
rows = (melt(df, id_vars=[], value_vars=categorical_cols) 
        .groupBy("variable", "value").count().collect()) 
 
mapping = {k: {x.value: x["count"] for x in v}  
          for k, v in groupby(sorted(rows), lambda x: x.variable)}

并使用udf来添加值:

from pyspark.sql.functions import udf 
 
def get_count(mapping_c): 
    @udf("bigint") 
    def _(x): 
        return mapping_c.get(x) 
    return _ 
 
 
for c in categorical_cols: 
    df = df.withColumn("{}_freq".format(c), get_count(mapping[c])(c))

你应该吗？或许。与迭代连接不同，它只需要一个 Action 来计算所有统计数据。如果结果较小(预期使用分类变量)，您可以获得适度的性能提升。

添加广播提示。

from pyspark.sql.functions import broadcast 
 
def calculate_freq(df, categorical_cols): 
    for each_cat_col in categorical_cols: 
        _freq = df.select(each_cat_col).groupBy(each_cat_col).count() 
    df = df.join(broadcast(_freq), each_cat_col, "inner") 
    return df

Spark 应该自动广播，所以它不应该改变任何事情，但帮助规划者总是更好。

Also, would it make a difference if I was simply running the above process in a loop and not inside a function?

忽略代码的可维护性和可测试性它不会。