Spark中清理数据的方法

Question

我是数据工程/机器学习和自学的新手。在处理示例问题时，我遇到了以下数据清理任务

1. Remove extra whitespaces (keep one whitespace in between word but remove more 
than one whitespaces) and punctuations

2. Turn all the words to lower case and remove stop words (list from NLTK)

3. Remove duplicate words in ASSEMBLY_NAME column

虽然我在大学作业期间一直致力于编写代码来执行这些任务，但我从来没有在一段代码中完成过（对于任何项目），我正在寻找可以提供帮助的专家的指导我指出了完成它的最佳方法(in python or scala)

目前已完成的工作：

1.从 parquet 文件中读取数据

partFitmentDF = sqlContext.read.parquet("/mnt/blob/devdatasciencesto/pga-parts-forecast/raw/parts-fits/")

display(partFitmentDF)

2。从 DF 创建表

partFitmentDF.createOrReplaceTempView("partsFits")
partFitmentDF.write.mode("overwrite").format("delta").saveAsTable("partsFitsTable")

3.重新排列表中 fit_assembly_name 的数据，以便所有 fit_assembly_name 和 fit_assembly_id 对于每个不同的 itemno 都滚动到单行

%sql

select itemno, concat_ws(' | ' , collect_set(cast(fits_assembly_id as int))) as fits_assembly_id, concat_ws(' | ' ,collect_set(fits_assembly_name)) as fits_assembly_name 
from partsFitsTable 
WHERE itemno = 1014584
group by itemno

附言

partFitmentDF 选定列的示例数据

itemno  fits_assembly_id        fits_assembly_name
0450056 44011           OIL PUMP ASSEMBLY - A01EA09CA (4999202399920239A06)
0450056 135502          OIL PUMP ASSEMBLY - A02EA09CA/CB/CC (4999202399920239A06)
0450056 37884           OIL PUMP ASSEMBLY - A01EA05CA (4999202399920239A06)
0450056 19618           OIL PUMP ASSEMBLY - A06FA09CA/CB/CC (4999202399920239A06)
0450056 135021          OIL PUMP ASSEMBLY - A02EA05CA (4999202399920239A06)
0450056 4147            OIL PUMP ASSEMBLY - A04KA05CA (4999202359920235A06)
0450056 12003           OIL PUMP ASSEMBLY - A05FA09CA/CB/CC (4999202399920239A06)

现在，我需要将这些多行按项目编号滚动为一行 （属于一个项目编号的所有 assembly_name 和 id 都应该在一行中） 然后我需要执行最顶部列出的任务#1、2 和 3 以清理 fit_assembly_name 列并将处理后的数据保存到带有 itemno、fits_assembly_id 和 fit_assembly_name 列的最终数据帧或表，但我不知道如何开始蟒蛇来做到这一点。您能否通过建议方法（和代码提示）帮助我，以便我可以进一步完成这项任务？

Answer 1

检查以下内容是否适合您。我假设 df 是您已经运行的 groupby 和 collect_set 之后的数据框：

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

任务1：使用RegexTokenizer

使用模式(?:\p{Punct}|\s)+ 分割字符串，将结果保存到temp1 列。生成的字符串数组将包含所有小写项目，前导/尾随空格也将被删除。

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

df1 = tk.transform(df)

任务2：使用StopWordsRemover

删除停用词并将结果保存到temp2 列：

sw = StopWordsRemover(inputCol='temp1', outputCol='temp2')

df2 = sw.transform(df1)

您可以通过输入 sw.getStopWords() 检查所有当前停用词，检查 loadDefaultStopWords(language) 以切换到另一种语言设置，或通过以下方式附加您自己的停用词：

mylist = sw.getStopWords() + ['my', 'black', 'list']
# then adjust the transformer to the following
sw = StopWordsRemover(inputCol='temp1', outputCol='temp2', stopWords=mylist)

此时，您应该有一个字符串数组列 temp2 已删除停用词。

任务 3：

使用array_distinct() 删除重复项，使用 concat_ws() 将数组转换为字符串，然后删除两个临时列：

df_new = df2.withColumn('fits_assembly_name', expr('concat_ws(" ", array_distinct(temp2))')) \
            .drop('temp1', 'temp2')

如果您对上述代码有任何问题，请告诉我。

Answer 2

您想使用 SQL 或 pyspark 函数。

• The pyspark docs

• The Spark SQL docs

• trim(s) -- 返回字符串s，带有前导和尾随空格

• regexp_replace(s, "\W") -- 返回字符串s，非字母数字字符（“标点符号”）被空格替换
• lower(s) -- 返回字符串s，所有字符都小写
• split(s) -- 在字符串s split 中返回一个单词数组，其中单词边界是一个或多个空格。
• array_distinct(a) -- 返回 a 的不同元素

在 pyspark 中像这样把它们放在一起：

from pyspark.sql import *
import pyspark.sql.functions as F

df = spark.createDataFrame( [Row(name="BODY, DECALS - Z19VF99LK/LE (702498)"), Row(name="     ABC  DEF ABC   ")] ) 
clean_df = df.withColumn("words", 
                         F.array_distinct(
                             F.split(
                               F.trim(
                                 F.regexp_replace(F.lower(df.name), "\W", " ")
                               )
                             , " +")
                         )
                        )
clean_df.show(truncate=False)

要生成带有结果的words 列：

+------------------------------------+-------------------------------------+
|name                                |words                                |
+------------------------------------+-------------------------------------+
|BODY, DECALS - Z19VF99LK/LE (702498)|[body, decals, z19vf99lk, le, 702498]|
|     ABC  DEF ABC                   |[abc, def]                           |
+------------------------------------+-------------------------------------+

我对确切的问题有点不清楚，但这应该会让你朝着正确的方向前进。