在 spark 2.4.4 中修复 assembly_id 和 assembly_name 列数据的方法答案

【问题标题】：Approach to fix assembly_id and assembly_name column data in spark 2.4.4在 spark 2.4.4 中修复 assembly_id 和 assembly_name 列数据的方法
【发布时间】：2020-02-27 23:22:29
【问题描述】：

我一直在 spark 2.4.4 中处理数据清理任务，但被困在以下两个任务中（在问题部分提到）。以下是数据框和问题详细信息：

1.在数据框中挂载数据并读取 parquet 文件

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

2。样本数据

display(partFitmentRawDF)

Itemno   Assembly_id    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)

我已经完成了其他处理以使数据看起来像上面，但我被困在以下任务中

3.问题

a. 如果我们查看 row2 和列 Assembly_name 则有三个 id 的 A02EA09CA/CB/CC 但它们已被合并。您能否建议如何制作它 A02EA09 A02EA09CB A02EA09CC。基本上，所有部分都应该有独立的 id 并在它们之间连接一个空格。同样问题的另一个例子是将DRIVE TRAIN, TRANSMISSION (6 SPEED) - V08AB26/GB26/LB26 ALL OPTIONS (49VICTRANS08)更改为DRIVE TRAIN, TRANSMISSION (6 SPEED) - V08AB26 V08GB26 V08LB26 ALL OPTIONS (49VICTRANS08)或将SUSPENSION (7043244) - S09PR6HSL/PS6HSL/HEL (49SNOWSHOCKFRONT7043244SB)更改为SUSPENSION (7043244) - S09PR6HSL S09PS6HSL S09PS6HEL (49SNOWSHOCKFRONT7043244SB)。

b.将assemble_id and assembly_name column中属于同一itemno的多行滚动到单行并删除重复的单词。

所以关注数据集

Itemno   Assembly_id    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)

将产生下面提到的最终数据集（没有标点符号，单词和重复单词之间有多个空格）

Itemno   Assembly_id            Assembly_name
0450056  44011 135502 37884     OIL PUMP ASSEMBLY A01EA09CA 4999202399920239A06 A02EA09CA A02EA09CB A02EA09CC 4999202399920239A06

您能帮我解决这个问题吗？提前感谢您的帮助！

测试JXC方案后的问题

1.连接问题

如果初始数据集如下

itemno  fits_assembly_id    fits_assembly_name
1322660 35459               DRIVE TRAIN, CLUTCH, PRIMARY - S09PR6HSL/PS6HSL/HEL (49SNOWDRIVECLUTCH09600TRG)

在step-3之后，它正在执行以下操作

+-------------------------------------------------------------------------------------------+
|temp1                                                                                      |
+-------------------------------------------------------------------------------------------+
|[DRIVE, TRAIN, CLUTCH, PRIMARY, S09PR6HSL, S09PS6HSL, S09PS6HEL, 49SNOWDRIVECLUTCH09600TRG]|

最初，它是S09PR6HSL/PS6HSL/HEL，所以它应该更改为S09PR6HSL S09PS6HSL S09PR6HEL，但它是S09PR6HSL S09PS6HSL S09PS6HEL。第三项应该是S09PR6HEL，但它是S09PS6HEL。我相信它应该从第一个字符串中连接部分并将其添加到/之后的所有其他部分。

2。下划线替换：这是新的，因为我刚刚发现了几行。有时/ 之后的字符串有下划线。在这种情况下，与下划线相同位置的字母 （第一个字符串） 应替换第一个字符串的第二个或后面的字符串中的 _。例如，如果数据是DRIVE TRAIN, CLUTCH, PRIMARY - S09PR6HSL/PS_HSL/H_L (49SNOWDRIVECLUTCH09600TRG)，那么它应该更改为DRIVE TRAIN, CLUTCH, PRIMARY - S09PR6HSL S09PS6HSL S09PR6HSL (49SNOWDRIVECLUTCH09600TRG)。在这里，In /PS_HSL/ _ 被值 6 替换，因为 PS_HSL 与 PR6HSL 匹配，因此将 _ 替换为 6 并添加 S09 以使其完整 id 为 S09PS6HSL。基本上，从第一个字符串中获取数据并将其附加到后面的字符串（在/ 之后），如果它的长度不同使其成为完整的 id。如果有_，则从第一个字符串中取与_ 相同位置的数据，并在后面的ID 字符串中替换它。

3. 分离由/ 连接的子字符串，并将其余字符串存储在添加到同一数据帧的新列中

例如：

step0后的问题：

【问题讨论】：

标签： python-3.x scala dataframe apache-spark pyspark

【解决方案1】：

好问题，理想情况下我会使用 udf 以使事情变得简单，但由于此任务是使用 Spark SQL 高阶函数的一个很好的例子......可能有点冗长，所以我把它分成4个步骤。让我知道它是否有效，欢迎提出任何问题：

Step-1：将字符串转换为字符串数组

用连续的模式(?:(?!/)\p{Punct}|\s)+'))分割字符串标点符号（/ 除外）或空格，然后过滤掉 EMPTY（前导/尾随）的项目。临时列temp1 用于保存中间列。

from pyspark.sql.functions import split, expr

df1 = df.withColumn('temp1', split('Assembly_name', r'(?:(?!/)\p{Punct}|\s)+')) \
        .withColumn('temp1', expr("filter(temp1, x -> x <> '')"))

df1.select('temp1').show(truncate=False)
+-------------------------------------------------------------------------------------+
|temp1                                                                                |
+-------------------------------------------------------------------------------------+
|[OIL, PUMP, ASSEMBLY, A01EA09CA, 4999202399920239A06]                                |
|[OIL, PUMP, ASSEMBLY, A02EA09CA/CB/CC, 4999202399920239A06]                          |
|[OIL, PUMP, ASSEMBLY, A01EA05CA, 4999202399920239A06]                                |
|[DRIVE, TRAIN, TRANSMISSION, 6, SPEED, V08AB26/GB26/LB26, ALL, OPTIONS, 49VICTRANS08]|
|[SUSPENSION, 7043244, S09PR6HSL/PS6HSL/HEL, 49SNOWSHOCKFRONT7043244SB]               |
+-------------------------------------------------------------------------------------+

Step-2：将 temp1 转换为数组数组

使用/再次拆分数组项，以便所有part-id都在自己的数组项上

df2 = df1.withColumn('temp1', expr("transform(temp1, x -> split(x, '/'))"))
df2.select('temp1').show(truncate=False)
+----------------------------------------------------------------------------------------------------------+
|temp1                                                                                                     |
+----------------------------------------------------------------------------------------------------------+
|[[OIL], [PUMP], [ASSEMBLY], [A01EA09CA], [4999202399920239A06]]                                           |
|[[OIL], [PUMP], [ASSEMBLY], [A02EA09CA, CB, CC], [4999202399920239A06]]                                   |
|[[OIL], [PUMP], [ASSEMBLY], [A01EA05CA], [4999202399920239A06]]                                           |
|[[DRIVE], [TRAIN], [TRANSMISSION], [6], [SPEED], [V08AB26, GB26, LB26], [ALL], [OPTIONS], [49VICTRANS08]] |
|[[SUSPENSION], [7043244], [S09PR6HSL, PS6HSL, HEL], [49SNOWSHOCKFRONT7043244SB]]                          |
+----------------------------------------------------------------------------------------------------------+

第 3 步：使用聚合重置部分 ID

聚合函数将对内部数组进行操作：

df3 = df2.withColumn('temp1', expr("""

       flatten(
         transform(temp1, x ->
           transform(sequence(1, size(x)), i ->
             aggregate(
                 sequence(1, i)
               , x[0]
               , (acc,j) -> concat(substr(acc, 1, length(x[0])-length(x[j-1])), x[j-1])
             )
           )
         )
       )

    """))

df3.select('temp1').show(truncate=False)
+----------------------------------------------------------------------------------------------+
|temp1                                                                                         |
+----------------------------------------------------------------------------------------------+
|[OIL, PUMP, ASSEMBLY, A01EA09CA, 4999202399920239A06]                                         |
|[OIL, PUMP, ASSEMBLY, A02EA09CA, A02EA09CB, A02EA09CC, 4999202399920239A06]                   |
|[OIL, PUMP, ASSEMBLY, A01EA05CA, 4999202399920239A06]                                         |
|[DRIVE, TRAIN, TRANSMISSION, 6, SPEED, V08AB26, V08GB26, V08LB26, ALL, OPTIONS, 49VICTRANS08] |
|[SUSPENSION, 7043244, S09PR6HSL, S09PS6HSL, S09PS6HEL, 49SNOWSHOCKFRONT7043244SB]             |
+----------------------------------------------------------------------------------------------+

地点：

transform(temp1, x -> func1(x)) : 遍历数组temp1中的每一项运行func1(x)，x是内部数组（字符串数组）
上面提到的
func1(x) 是另一个变换函数，它遍历 sequence(1, size(x)) 并在每个 上运行 func2(i)我：
```
transform(sequence(1, size(x)), i -> func2(i))
```
上面提到的
func2(i)是一个聚合函数，它遍历sequence(1,i)，初始值为x[0]，并累加/减少使用函数：
```
(acc,j) -> concat(substr(acc, 1, length(acc)-length(x[j-1])), x[j-1])
```
注意： substr() 的位置是从 1 开始的，array-indexing 是从 0 开始的，因此我们需要 x[j-1] 来引用上面 reduce/ 中的当前数组项聚合函数

最后，运行flatten() 合并内部数组

这一步是在做类似下面的 pysudo-loop：

for x in temp1:
  for i in range(1, size(x)+1):
    acc = x[0]
    for j in range(1,i+1):
      acc = concat(substr(acc, 1, length(acc)-length(x[j-1])), x[j-1])

第 4 步：合并和删除重复项

df4 = df3.groupby('Itemno').agg(
      expr("concat_ws(' ', array_distinct(flatten(collect_list(temp1)))) AS Assembly_names")
    , expr("concat_ws(' ', collect_set(Assembly_id)) as Assembly_ids")
  )

地点：

使用collect_list()获取数组数组（temp1是字符串数组）
使用flatten()将上面的转换成字符串数组
使用 array_distinct() 删除重复项

使用concat_ws()将上面的数组转换成字符串

df4.select('Assembly_names').show(truncate=False)
+---------------------------------------------------------------------------------------+
|Assembly_names                                                                         |
+---------------------------------------------------------------------------------------+
|OIL PUMP ASSEMBLY A01EA09CA 4999202399920239A06 A02EA09CA A02EA09CB A02EA09CC A01EA05CA|
|SUSPENSION 7043244 S09PR6HSL S09PS6HSL S09PS6HEL 49SNOWSHOCKFRONT7043244SB             |
|DRIVE TRAIN TRANSMISSION 6 SPEED V08AB26 V08GB26 V08LB26 ALL OPTIONS 49VICTRANS08      |
+---------------------------------------------------------------------------------------+

更新：

第一个很容易修复，它比现有的要容易得多（不需要聚合）。对于第二个，以下解决方案需要逐个字符地遍历字符串，这可能会很慢。如果是这样，我们可能必须使用 udf 进行检查。

以下是变化：

第 1 步：只需添加下划线即可从标点符号中排除：（请注意，如果在文本的其他位置显示任何下划线，可能需要先运行 regexp_replace 以清理它们）
```
df1 = df.withColumn('temp1', split('Assembly_name', r'(?:(?![/_])\p{Punct}|\s)+')) \ 
        .withColumn('temp1', expr("filter(temp1, x -> x <> '')"))
```

Step-2： 将数组数组进一步拆分为数组数组数组，最里面的数组将字符串拆分为字符。反转最里面的数组，以便于比较。

df2 = df1.withColumn('temp1', expr("transform(temp1, x -> split(x, '/'))")) \
         .withColumn('temp1', expr("transform(temp1, x -> transform(x, y -> reverse(split(y, ''))) )"))

第 3 步： 使用 transform() 而不是 aggregate() 来重置零件 ID。我们检查 y[i]（最内层数组的项）是否为 NULL 或下划线，然后将其替换为 x[0][i]中的相应项>。然后我们反转数组并使用 concat_ws(''..) 将其转换回字符串。

df3 = df2.withColumn('temp1', expr("""

   flatten(
     transform(temp1, x ->
       transform(x, y ->
         concat_ws('', 
           reverse(
             transform(sequence(0, size(x[0])-1), i -> IF(y[i] is NULL or y[i] == '_', x[0][i], y[i]))
           )
         )
       )
     ) 
   ) 

"""))

下面是上面的结果

df3.select('temp1').show(truncate=False)                                                                           
+---------------------------------------------------------------------------------------------+
|temp1                                                                                        |
+---------------------------------------------------------------------------------------------+
|[OIL, PUMP, ASSEMBLY, A01EA09CA, 4999202399920239A06]                                        |
|[OIL, PUMP, ASSEMBLY, A02EA09CA, A02EA09CB, A02EA09CC, 4999202399920239A06]                  |
|[OIL, PUMP, ASSEMBLY, A01EA05CA, 4999202399920239A06]                                        |
|[DRIVE, TRAIN, TRANSMISSION, 6, SPEED, V08AB26, V08GB26, V08LB26, ALL, OPTIONS, 49VICTRANS08]|
|[SUSPENSION, 7043244, S09PR6HSL, S09PS6HSL, S09PR6HEL, 49SNOWSHOCKFRONT7043244SB]            |
|[DRIVE, TRAIN, CLUTCH, PRIMARY, S09PR6HSL, S09PS6HSL, S09PR6HSL, 49SNOWDRIVECLUTCH09600TRG]  |
|[DRIVE, TRAIN, CLUTCH, PRIMARY, S09PR6HSL, S09PS6HSL, S09PR6HSL, 49SNOWDRIVECLUTCH09600TRG]  |
+---------------------------------------------------------------------------------------------+

处理前的字段：

df.select('Assembly_name').show(truncate=False)                                                                    
+----------------------------------------------------------------------------------+
|Assembly_name                                                                     |
+----------------------------------------------------------------------------------+
|OIL PUMP ASSEMBLY - A01EA09CA (4999202399920239A06)                               |
|OIL PUMP ASSEMBLY - A02EA09CA/CB/CC (4999202399920239A06)                         |
|OIL PUMP ASSEMBLY - A01EA05CA (4999202399920239A06)                               |
|DRIVE TRAIN, TRANSMISSION (6 SPEED) - V08AB26/GB26/LB26 ALL OPTIONS (49VICTRANS08)|
|SUSPENSION (7043244) - S09PR6HSL/PS6HSL/HEL (49SNOWSHOCKFRONT7043244SB)           |
|DRIVE TRAIN, CLUTCH, PRIMARY - S09PR6HSL/PS_HSL/H_L (49SNOWDRIVECLUTCH09600TRG)   |
|DRIVE TRAIN, CLUTCH, PRIMARY - S09PR6HSL/_S__SL/H_L (49SNOWDRIVECLUTCH09600TRG)   |
+----------------------------------------------------------------------------------+

第 4 步：没有变化。

UPDATE-2 添加了 Step-0：

Step-0： 预处理列Assembly_name，使用regexp_replace + split 将模型# 分离成一个新列并从原列Assembly_name中删除：

from pyspark.sql.functions import regexp_replace, split

df0 = df.withColumn('new_col', split(regexp_replace('Assembly_name', r'^(.*)-\s*(\S+)(.*)$', '$1$3\0$2'),'\0')) \
    .selectExpr(
        'Itemno'
      , 'Assembly_id'
      , "coalesce(new_col[0], Assembly_name) as Assembly_name"
      , "coalesce(new_col[1], '') as models"
)

df0.show(truncate=False)
+-------+-----------+---------------------------------------------------------------+--------------------+
|Itemno |Assembly_id|Assembly_name                                                  |models              |
+-------+-----------+---------------------------------------------------------------+--------------------+
|0450056|44011      |OIL PUMP ASSEMBLY  (4999202399920239A06)                       |A01EA09CA           |
|0450056|135502     |OIL PUMP ASSEMBLY  (4999202399920239A06)                       |A02EA09CA/CB/CC     |
|0450056|37884      |OIL PUMP ASSEMBLY  (4999202399920239A06)                       |A01EA05CA           |
|0450067|12345      |DRIVE TRAIN, TRANSMISSION (6 SPEED)  ALL OPTIONS (49VICTRANS08)|V08AB26/GB26/LB26   |
|0450068|1000       |SUSPENSION (7043244)  (49SNOWSHOCKFRONT7043244SB)              |S09PR6HSL/PS6HSL/HEL|
|0450066|12345      |DRIVE TRAIN, CLUTCH, PRIMARY  (49SNOWDRIVECLUTCH09600TRG)      |S09PR6HSL/PS_HSL/H_L|
|0450069|12346      |DRIVE TRAIN, CLUTCH, PRIMARY (49SNOWDRIVECLUTCH09600TRG)       |                    |
+-------+-----------+---------------------------------------------------------------+--------------------+

然后您可以使用 RegexTokenier 和 StopwordsRemover 处理 Assembly_name，models 是当前帖子的简化版本，您可以跳过第 1 步，但请注意数组的深度。

（注意：从最后一条记录中删除S09PR6HSL/_S__SL/H_L进行测试）

【讨论】：

我可以问最后一个基本问题吗？我知道\p{punct} 类，但我不确定(?:(?!/)\p{Punct}|\s)+')) 中的所有部分是什么意思？我知道在java中使用[\p{Punct}&&[^()]]，这意味着除了(和)之外的所有标点符号。我有一个想法，但我不确定(?:(?!/)\p{Punct}|\s)+')) 中的(?:(?!/) 和\s)+'))。你能解释一下吗？
嗨，@SachinSharma 检查regular-expressions.info/posixbrackets.html，\p{Punct} 的最后一列，您可以找到包含斜杠的相应 ASCII，因此我们在前面加上一个否定的前瞻 (?!/) 以排除斜杠。 (?:A|B) 是设置非捕获组，以便我们可以在该组的一个或多个匹配上实现+。在这个组中，A 是 (?!/)\p{Punct}，它匹配除斜杠以外的任何标点符号，B 是 \s，它是空格。
@SachinSharma，这个最像模型后面的一个hython#，你能把模式调整为：r'^(.*?)-\s*(\S+)(.*)$'
只是一个后续，如果你想认真学习正则表达式及其工作原理，你需要阅读这本书：amazon.com/Mastering-Regular-Expressions-Jeffrey-Friedl/dp/…
或检查stackoverflow中的#regex标签