在大数据上优化 Pyspark UDF答案

【问题标题】：Optimizing Pyspark UDF on large data在大数据上优化 Pyspark UDF
【发布时间】：2021-05-22 12:34:50
【问题描述】：

我正在尝试优化此代码，该代码在（pyspark 数据帧的）列值位于 [categories] 中时创建一个虚拟对象。

当运行 100K 行时，运行大约需要 30 秒。就我而言，我有大约 2000 万行，这将花费大量时间。

def create_dummy(dframe,col_name,top_name,categories,**options):
    lst_tmp_col = []
    if 'lst_tmp_col' in options:
        lst_tmp_col = options["lst_tmp_col"]
    udf = UserDefinedFunction(lambda x: 1 if x in categories else 0, IntegerType())
    dframe  = dframe.withColumn(str(top_name), udf(col(col_name))).cache()
    dframe = dframe.select(lst_tmp_col+ [str(top_name)])
    return dframe

换句话说，我如何优化此功能并减少关于我的数据量的总时间？以及如何确保这个函数不会遍历我的数据？

感谢您的建议。谢谢

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql user-defined-functions

【解决方案1】：

您不需要 UDF 来对类别进行编码。你可以使用isin:

import pyspark.sql.functions as F

def create_dummy(dframe, col_name, top_name, categories, **options):
    lst_tmp_col = []
    if 'lst_tmp_col' in options:
        lst_tmp_col = options["lst_tmp_col"]
    dframe = dframe.withColumn(str(top_name), F.col(col_name).isin(categories).cast("int")).cache()
    dframe = dframe.select(lst_tmp_col + [str(top_name)])
    return dframe

【讨论】：