Pyspark：通过搜索字典替换列中的值

Question

我是 PySpark 的新手。

我有一个DataFrame df包含“device_type”列的 Spark 。

我想将“平板电脑”或“电话”中的每个值替换为“电话”，并将“PC”替换为“桌面”。

在 Python 中，我可以执行以下操作，

deviceDict = {'Tablet':'Mobile','Phone':'Mobile','PC':'Desktop'}
df['device_type'] = df['device_type'].replace(deviceDict,inplace=False)

如何使用 PySpark 实现这一目标？谢谢！

Answer 1

您可以使用na.replace：

df = spark.createDataFrame([
    ('Tablet', ), ('Phone', ),  ('PC', ), ('Other', ), (None, )
], ["device_type"])

df.na.replace(deviceDict, 1).show()

df = spark.createDataFrame([
    ('Tablet', ), ('Phone', ),  ('PC', ), ('Other', ), (None, )
], ["device_type"])

df.na.replace(deviceDict, 1).show()

或地图文字：

from itertools import chain
from pyspark.sql.functions import create_map, lit

mapping = create_map([lit(x) for x in chain(*deviceDict.items())])


df.select(mapping[df['device_type']].alias('device_type'))

+-----------+
|device_type|
+-----------+
|     Mobile|
|     Mobile|
|    Desktop|
|      Other|
|       null|
+-----------+

请注意，后一种解决方案会将映射中不存在的值转换为NULL. 如果这不是您想要的行为，您可以添加coalesce：

from pyspark.sql.functions import coalesce


df.select(
    coalesce(mapping[df['device_type']], df['device_type']).alias('device_type')
)

from itertools import chain
from pyspark.sql.functions import create_map, lit

mapping = create_map([lit(x) for x in chain(*deviceDict.items())])


df.select(mapping[df['device_type']].alias('device_type'))

Answer 2

经过大量搜索和替代方案后，我认为使用 python dict 替换的最简单方法是使用 pyspark dataframe 方法replace：

deviceDict = {'Tablet':'Mobile','Phone':'Mobile','PC':'Desktop'}
df_replace = df.replace(deviceDict,subset=['device_type'])

这将用 dict 替换所有值，df.na.replace()如果您将 dict 参数与子集参数相结合，您可以获得相同的结果 。这不是他的足够清晰的文档，因为如果你搜索的功能，replace你会得到两个引用，其中一个里面pyspark.sql.DataFrame.replace，另一个在一边pyspark.sql.DataFrameNaFunctions.replace，但两者参考使用的示例代码df.na.replace，因此不清楚，你可以实际使用df.replace。

Answer 3

您也可以使用以下方法执行此操作df.withColumn：

from itertools import chain
from pyspark.sql.functions import create_map, lit

deviceDict = {'Tablet':'Mobile','Phone':'Mobile','PC':'Desktop'}

mapping_expr = create_map([lit(x) for x in chain(*deviceDict.items())])

df = df.withColumn('device_type', mapping_expr[df['dvice_type']])
df.show()

Answer 4

这是一个受 Rrecode函数启发的小辅助函数，它抽象了之前的答案。作为奖励，它添加了默认值的选项。

from itertools import chain
from pyspark.sql.functions import col, create_map, lit, when, isnull
from pyspark.sql.column import Column

df = spark.createDataFrame([
    ('Tablet', ), ('Phone', ),  ('PC', ), ('Other', ), (None, )
], ["device_type"])

deviceDict = {'Tablet':'Mobile','Phone':'Mobile','PC':'Desktop'}

df.show()
+-----------+
|device_type|
+-----------+
|     Tablet|
|      Phone|
|         PC|
|      Other|
|       null|
+-----------+

这里是 的定义recode。

def recode(col_name, map_dict, default=None):
    if not isinstance(col_name, Column): # Allows either column name string or column instance to be passed
        col_name = col(col_name)
    mapping_expr = create_map([lit(x) for x in chain(*map_dict.items())])
    if default is None:
        return  mapping_expr.getItem(col_name)
    else:
        return when(~isnull(mapping_expr.getItem(col_name)), mapping_expr.getItem(col_name)).otherwise(default)

创建一个没有默认值的列会在所有不匹配的值中给出null/ None。

df.withColumn("device_type", recode('device_type', deviceDict)).show()

+-----------+
|device_type|
+-----------+
|     Mobile|
|     Mobile|
|    Desktop|
|       null|
|       null|
+-----------+

另一方面，为 指定一个default值会用这个默认值替换所有不匹配的值。

df.withColumn("device_type", recode('device_type', deviceDict, default='Other')).show()

+-----------+
|device_type|
+-----------+
|     Mobile|
|     Mobile|
|    Desktop|
|      Other|
|      Other|
+-----------+

Answer 5

最简单的方法是udf在您的数据框上应用 a ：

    from pyspark.sql.functions import col , udf

    deviceDict = {'Tablet':'Mobile','Phone':'Mobile','PC':'Desktop'}
    map_func = udf(lambda row : deviceDict.get(row,row))
    df = df.withColumn("device_type", map_func(col("device_type")))