我继承了一个程序，它修改pysparkDataframe中的一些字符串。其中一个步骤涉及到从字符串中的一些单词中删除双/三/etc字母，以及一个额外的异常列表，即使它们有重复的字母，这些异常也会被单独保留。目前，这是通过使用udf将Dataframe转换为pandas来完成的，然后在读回pyspark之前，对结果pandasDataframe中的字符串应用自定义函数。不幸的是，对需求的更改意味着代码在任何情况下都不能使用udf或Map到rdd。我需要直接在pyspark中执行相同的功能。
连续字符删除函数逐字读取字符串，检查单词是否在例外列表中，如果不在例外列表中，则逐字符移动，将其与上一个字符进行比较，如果存在匹配项，则逐字符检查是否创建新单词，省略重复。
下面是在pysparkDataframe转换为pandas之后，当前实现的mwe。

import pandas as pd

exception_list = ['ACCOUNTING', 'LOOK', 'FOOOOO']

cols = ['input']
data = [
    ["BOOK TOOK LOOK HOUSE SHOOK"],
    ["ACCOUNTING SHEEP"],
    ["FOO FOOO FOOOO FOOOOO FOOOOOO"]
]

df = pd.DataFrame(data, columns=cols)
df.head()

def drop_consecutive_chars(phrase, exception):
    if phrase == '':
        return phrase
    else:
        new_phrase = []
        for word in phrase.split():
            if word not in exception:
                prev = word[0]
                new_word = prev
                for char in word[1:]:
                    if char != prev:
                        new_word += char
                    prev = char
            else:
                new_word = word
            new_phrase += [new_word]
        new_phrase = ' '.join(new_phrase)
        return new_phrase

df['output'] = df['input'].apply(drop_consecutive_chars,
                                 exception=exception_list)

df.head()

在Pypark有什么办法吗？我对使用regextokenizer之类的东西持开放态度，稍后再加入它，并创建附加的truth列，这些列将在之后删除。它只需要在Dataframe没有离开pyspark或者被Map到其他任何东西的情况下完成。