Pyspark 过滤器使用列表中的startswith

Posted 2023-04-17

【中文标题】Pyspark 过滤器使用列表中的startswith

【英文标题】：Pyspark filter using startswith from list

【发布时间】：2018-01-31 18:43:25

【问题描述】：

我有一个元素列表，这些元素可能以 RDD 中记录的几个字符串开头。如果我有 yes 和 no 的元素列表，它们应该匹配 yes23 和 no3 但不匹配 35yes 或 41no。使用 pyspark，我如何使用startswith 列表或元组中的任何元素。

一个示例 DF 将是：

+-----+------+
|index| label|
+-----+------+
|    1|yes342|
|    2| 45yes|
|    3| no123|
|    4|  75no|
+-----+------+

当我尝试时：

Element_List = ['yes','no']
filter_DF = DF.where(DF.label.startswith(tuple(Element_List)))

生成的 df 应该类似于：

+-----+------+
|index| label|
+-----+------+
|    1|yes342|
|    3| no123|
+-----+------+

相反，我得到了错误：

Py4JError: An error occurred while calling o250.startsWith. Trace:
py4j.Py4JException: Method startsWith([class java.util.ArrayList]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
at py4j.Gateway.invoke(Gateway.java:272)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:745)

被提示，所以看起来startsWith 不能与任何类型的列表一起使用。有没有简单的解决方法？

【参考方案1】：

这样写表达式：

from pyspark.sql.functions import col, lit
from functools import reduce

element_list = ['yes','no']

df = spark.createDataFrame(
    ["yes23", "no3", "35yes", """41no["maybe"]"""],
    "string"
).toDF("location")

starts_with = reduce(
    lambda x, y: x | y,
    [col("location").startswith(s) for s in element_list], 
    lit(False))

df.where(starts_with).show()
# +--------+
# |location|
# +--------+
# |   yes23|
# |     no3|
# +--------+

【讨论】：

注意：此语法还支持负过滤，即 df.where(~starts_with) 将选择不以 element_list 中的元素开头的项目。

【参考方案2】：

我觉得最好的方法是使用原生 pyspark 函数，如“rlike()”。 startswith() 用于过滤静态字符串。 它不能接受动态内容。如果您想从列表中动态获取关键字；最好的办法是从下面的列表中创建一个正则表达式。

# List
li = ['yes', 'no']
# frame RegEx from the List
# in this case strings starting with yes/no i.e. "^(yes|no)"
reg_str = r"^("+ "|".join(li) + ")"

自定义 Udfs 或使用 RDD 函数可能效果很好，但使用自定义 Udfs 可能会影响性能。

下面是完整的工作示例。

#Test Dataframe
df = spark.createDataFrame(
    ["yes23", "no3", "35yes"],
    "string"
).toDF("label")

# List
li = ['yes', 'no']
# frame RegEx from the List
# in this case strings starting with yes/no i.e. "^(yes|no)"
reg_str = r"^("+ "|".join(li) + ")"

#Filter dataframe with RegEx
df.filter(df.label.rlike(reg_str)).show()

# +--------+
# |label   |
# +--------+
# |   yes23|
# |     no3|
# +--------+