PYSPARK如何使用dict创建UDF，然后使用UDF向 Dataframe 添加列

inn6fuwd 于 2022-11-21 发布在 Spark

关注(0)|答案(2)|浏览(141)

我需要在 pyspark 中创建一个 UDF 来转换字母等级（ " A " 、 " B " 、 " C " 、 " D " 、 " F " ）到数字等级（ 4 、 3 、 2 、 1 和 0 ）。然后我需要将此函数注册为 spark UDF 。接下来，我有一个数据帧 " current _ gpa " 。 Current _ gpa 有一个名为 " grade " 的列。我需要向数据帧 current _ gpa 添加一个名为 " num _ grade " 的列其中列 " grade " 中的字母等级被转换为列 " num _ grade " 中的相应数字。
这是我创建的 UDF ：

def get_num(letter):
    letter_class_dict = {"A": 1, "B": 2, "C": 3, "D": 4, "F": 5}
    for letter, l in letter_class_dict():
        x['letter'] = l
 
    return l

get_num =  udf(lambda letter: letter_class_dict.get(letter))
get_num_udf = F.udf(get_num, IntegerType())

中的每一个
这是数据帧 current _ gpa ：

+-------+-------+------+----+-----+-------+
| course|term_id|   sid| fid|grade|credits|
+-------+-------+------+----+-----+-------+
|BIO 101|  2000B|100001|1007|    F|      3|
|BIO 102|  2000B|100001|1007|    F|      4|
|CHM 101|  2000B|100001|1002|    F|      4|
|BIO 103|  2000B|100001|1007|    F|      4|
|GEN 114|  2000B|100001|1006|    F|      3|
+-------+-------+------+----+-----+-------+

格式
我尝试使用此 UDF 添加一个列 ' num _ grade ' ，其中的值应如下所示：
第一次
这会产生错误：UDF 引发了异常错误：' 执行阶段错误：SparkContext 只能在驱动程序上创建和访问。 " 。完整追溯如下：

pyspark

来源：https://stackoverflow.com/questions/74346960/pyspark-how-to-create-a-udf-with-a-dict-and-then-add-a-column-to-a-dataframe-usi

2条答案

按热度按时间

ivqmmu1c1#

您不需要UDF来执行此操作，并且您应该始终尽量避免UDF（除非绝对必要），因为Spark无法优化它们，这可能会导致性能下降。
这是一个简单的情况，当（when().otherwise()）操作可以使用列表解析中的字典项或python的本地map函数来构建时。

letter_class_dict = {"A": 4, "B": 3, "C": 2, "D": 1, "F": 0}

# create individual case when statement for each swap
letter_class_casewhens = map(lambda a: func.when(func.col('grade') == a[0], func.lit(a[1])), 
                             letter_class_dict.items()
                             )

# [Column<'CASE WHEN (grade = A) THEN 4 END'>,
#  Column<'CASE WHEN (grade = B) THEN 3 END'>,
#  Column<'CASE WHEN (grade = C) THEN 2 END'>,
#  Column<'CASE WHEN (grade = D) THEN 1 END'>,
#  Column<'CASE WHEN (grade = F) THEN 0 END'>]

# pass the case when statements in a `coalesce` function
data_sdf. \
    withColumn('num_grades', func.coalesce(*letter_class_casewhens)). \
    show()

# +-------+-------+------+----+-----+-------+----------+
# | course|term_id|   sid| fid|grade|credits|num_grades|
# +-------+-------+------+----+-----+-------+----------+
# |BIO 101|  2000B|100001|1007|    F|      3|         0|
# |BIO 102|  2000B|100001|1007|    F|      4|         0|
# |CHM 101|  2000B|100001|1002|    F|      4|         0|
# |BIO 103|  2000B|100001|1007|    F|      4|         0|
# |GEN 114|  2000B|100001|1006|    F|      3|         0|
# +-------+-------+------+----+-----+-------+----------+

赞(0）回复(0）举报 2022-11-21

5gfr0r5j2#

下面是我创建UDF将字母等级转换为数字的过程：

def convert_grades(letter):
    letter_grades = {
    'A':4,
    'B': 3,
    'C':2,
    'D':1,
    'F':0
  }
    return letter_grades.get(letter)
 
grade_points = spark.udf.register('convert_grades', convert_grades)

赞(0）回复(0）举报 2022-11-21

我来回答

PYSPARK如何使用dict创建UDF，然后使用UDF向 Dataframe 添加列

2条答案

相关问题

热门标签

最新问答