python - 如何在pySpark中从字符串数据框中有效替换多个正则表达式模式的所有实例？

Question

我在Hadoop中有一个表，其中包含70亿个字符串，这些字符串本身可以包含任何内容。我需要从包含字符串的列中删除每个名称。一个示例字符串将是“约翰去公园”，我需要从中删除“约翰”，理想情况下只需替换为“[名称]”。

在“约翰和玛丽进入市场”的情况下，输出为“[NAME]，[NAME]进入市场”。

为此，我列出了最常见的20k名称。

我可以访问Hue(Hive，Impala)和Zeppelin(Spark，Python和库)来执行此操作。

我已经在数据库中进行了尝试，但是由于无法更新列或遍历变量而使其无法使用，因此使用Python和PySpark似乎是最好的选择，尤其是考虑到计算数量(20k名称* 70亿输入)字符串)

#nameList contains ['John','Emma',etc]
def removeNames(line, nameList):
    str_line= line[0]
    for name in nameList:
        rx = f"(^| |[[:^alpha:]])({name})( |$|[[:^alpha:]])"
        str_line = re.sub(rx,'[NAME]', str_line)
    str_line= [str_line]
    return tuple(str_line)

df = session.sql("select free_text from table")
rdd = df.rdd.map(lambda line: removeNames(line, nameList))
rdd.toDF().show()

代码正在执行，但是即使我将输入文本限制为1000行(这对于Spark来说也不是)，这也要花一个半小时，并且最终输出中实际上并未替换这些行。

我想知道的是:为什么 map 实际上不更新RDD的行，我如何使它更有效，使其在合理的时间内执行？

这是我的第一次发布，因此，如果缺少必要的信息，我将尽力填写。

谢谢!

Answer 1

如果您仍然对此感到好奇，通过使用udf(您的removeNames函数)，Spark会将所有数据序列化到主节点，从根本上击败了您使用Spark进行分布式操作。正如注释中建议的方法一样，如果您使用regexp_replace()方法，Spark将能够将所有数据保留在分布式节点上，从而使所有数据保持分布式并提高性能。