Примените StopWordsRemover и RegexTokenizer к нескольким столбцам в Spark 2.4.3. - PullRequest
Новая
       4

Примените StopWordsRemover и RegexTokenizer к нескольким столбцам в Spark 2.4.3.

1 голос
/ 05 ноября 2019

У меня есть следующий фрейм данных, df4 

|Itemno   |fits_assembly_id                                        |fits_assembly_name                                                                         |assembly_name 

|0450056  |13039 135502 141114 4147 138865 2021 9164               |OIL PUMP ASSEMBLY A01EA09CA 4999202399920239A06 A02EA09CA A02EA09CB A02EA09CC              |OIL PUMP ASSEMBLY 999202399920239A06

, и я использую следующий код для обработки / очистки вышеупомянутого фрейма данных 

from pyspark.ml.feature import StopWordsRemover, RegexTokenizer
from pyspark.sql.functions import expr


# Task-1: Regex Tokenizer

tk = RegexTokenizer(pattern=r'(?:\p{Punct}|\s)+', inputCol='fits_assembly_name', outputCol='temp1')
df5 = tk.transform(df4)

#Task-2: StopWordsRemover
sw = StopWordsRemover(inputCol='temp1', outputCol='temp2')
df6 = sw.transform(df5)

# #Task-3: Remove duplicates
df7 = df6.withColumn('fits_assembly_name', expr('concat_ws(" ", array_distinct(temp2))')) \
            .drop('temp1', 'temp2')

Я хочу обработать обастолбцы fits_assembly_name и assembly_name в RegexTokenizer & StopWordsRemover за один раз. Не могли бы вы поделиться, как это может быть достигнуто?

1 Ответ

1 голос
/ 05 ноября 2019

Вы можете использовать понимание списка для обработки нескольких столбцов, используйте pyspark.ml.Pipeline , чтобы пропустить промежуточные кадры данных, см. Ниже: 

from pyspark.ml.feature import StopWordsRemover, RegexTokenizer
from pyspark.ml import Pipeline
from pyspark.sql.functions import expr

# df4 is the initial dataframe and new result will overwrite it.
for col in ['fits_assembly_name', 'assembly_name']:
    tk = RegexTokenizer(pattern=r'(?:\p{Punct}|\s)+', inputCol=col, outputCol='temp1')
    sw = StopWordsRemover(inputCol='temp1', outputCol='temp2')
    pipeline = Pipeline(stages=[tk, sw])
    df4 = pipeline.fit(df4).transform(df4) \
        .withColumn(col, expr('concat_ws(" ", array_distinct(temp2))')) \
        .drop('temp1', 'temp2')
...