Извлечь несколько слов, используя regexp_extract в pyspark

Question

У меня есть список, который содержит несколько слов, и мне нужно извлечь соответствующие слова из текстовой строки, я нашел this , но он извлекает только одно слово

файл ключейcontent

это ключевое слово

part_description file content

32015 это ключевое слово hello world

код

import pyspark.sql.functions as F

keywords = sc.textFile('file:///home/description_search/keys') #1
part_description =  sc.textFile('file:///description_search/part_description') #2
keywords = keywords.map(lambda x: x.split(' ')) #3
keywords = keywords.collect()[0] #4
df = part_description.map(lambda r: Row(r)).toDF(['line']) #5
df.withColumn('extracted_word', F.regexp_extract(df['line'],'|'.join(keywords), 0)).show() #6

выходы

+--------------------+--------------+
|                line|extracted_word|
+--------------------+--------------+
|32015   this is a...|          this|
+--------------------+--------------+

Ожидаемый результат

+--------------------+-----------------+
|                line|   extracted_word|
+--------------------+-----------------+
|32015   this is a...|this,is,a,keyword|
+--------------------+-----------------+

Я хочу

1. возвращает все подходящие ключевые слова и их количество

2. , и если step #4 является наиболее эффективным способом

Воспроизводимый пример:

keywords = ['this','is','a','keyword']
l = [('32015 this is a keyword hello world'      , ),
('keyword this'      ,   ),
('32015 this is a keyword hello world 32015 this is a keyword hello world'      ,   ),
('keyword keyword'      ,   ),
('is a'      , )]

columns = ['line']

df=spark.createDataFrame(l, columns)

Answer 1

Мне удалось решить с помощью UDF вместо этого, как показано ниже

def build_regex(keywords):
    res = '('
    for key in keywords:
        res += '\\b' + key + '\\b|'
    res = res[0:len(res) - 1] + ')'

    return res


def get_matching_string(line, regex):
    matches = re.findall(regex, line)
    return matches if matches else None


udf_func = udf(lambda line, regex: get_matching_string(line, regex),
               ArrayType(StringType()))

df = df.withColumn('matched', udf_func(df['line'], F.lit(build_regex(keywords)))).withColumn('count', F.size('matched'))

Результат

+--------------------+--------------------+-----+
|                line|             matched|count|
+--------------------+--------------------+-----+
|32015    this is ...|[this, is, this, ...|    5|
|12832    Shb is a...|             [is, a]|    2|
|35015    this is ...|          [this, is]|    2|
+--------------------+--------------------+-----+