使用命名正则表达式从 PySpark DataFrame 中的列中提取多列

Question

假设我在 pySpark 中有一个 DataFrame df，格式如下：

| id | type | description                                                  |
|  1 | "A"  | "Date: 2018/01/01\nDescr: This is a test des\ncription\n     |
|  2 | "B"  | "Date: 2018/01/02\nDescr: Another test descr\niption\n       |
|  3 | "A"  | "Date: 2018/01/03\nWarning: This is a warnin\ng, watch out\n |

这当然是一个虚拟集合，但对于这个例子来说就足够了。

我已经做了一个带有命名组的正则表达式语句，可用于从描述字段中提取相关信息，类似于：

^(?:(?:Date: (?P<DATE>.+?)\n)|(?:Descr: (?P<DESCR>.+?)\n)|(?:Warning: (?P<WARNING>.+?)\n)+$

再次，虚拟正则表达式，实际的正则表达式稍微复杂一些，但目的是捕获三个可能的组：

| DATE       | DESCR                        | WARNING                        |
| 2018/01/01 | This is a test des\ncription | None                           |
| 2018/01/02 | Another test descr\niption   | None                           |
| 2018/01/03 | None                         | This is a warnin\ng, watch out |

现在我想将作为正则表达式匹配结果的列添加到原始 DataFrame（即将这个问题中的两个虚拟表合并为一个）。

我已经尝试了几种方法来实现这一点，但还没有一种方法可以提供完整的解决方案。我尝试过的一件事是：

def extract_fields(string):
   patt = <ABOVE_PATTERN>
   result = re.match(patt, string, re.DOTALL).groupdict()
   # Actually, a slight work-around is needed to overcome the None problem when 
   #   no match can be made, I'm using pandas' .str.extract for this now
   return result

df.rdd.map(lambda x: extract_fields(x.description))

这将产生第二个表，但我认为无法将其与来自df 的原始列结合起来。我试图构建一个新的Row()，但后来我遇到了@ 987654329@-constructor，导致数据框的列全部混乱。我怎样才能实现我想要的，即一个包含六列的 DataFrame：id、type、description、DATE、DESCR 和 WARNING？

备注。实际上，描述字段不仅仅是一个字段，而是几列。使用concat_ws，我将这些列连接成一个新列description，其中描述字段用\n 分隔，但也许可以以更好的方式合并。

Answer 1

我认为您可以在这种情况下使用 Pandas 功能。首先，我将 df 转换为 rdd 以拆分描述字段。我拉出 Pandas df，然后使用 Pandas df 创建 spark df。无论描述字段中的列号如何，它都可以工作

>>> import pandas as pd
>>> import re
>>> 
>>> df.show(truncate=False)
+---+----+-----------------------------------------------------------+
|id |type|description                                                |
+---+----+-----------------------------------------------------------+
|1  |A   |Date: 2018/01/01\nDescr: This is a test des\ncription\n    |
|2  |B   |Date: 2018/01/02\nDescr: Another test desc\niption\n       |
|3  |A   |Date: 2018/01/03\nWarning: This is a warnin\ng, watch out\n|
+---+----+-----------------------------------------------------------+

>>> #convert df to rdd
>>> rdd = df.rdd.map(list)
>>> rdd.first()
[1, 'A', 'Date: 2018/01/01\\nDescr: This is a test des\\ncription\\n']
>>> 
>>> #split description field
>>> rddSplit = rdd.map(lambda x: (x[0],x[1],re.split('\n(?=[A-Z])', x[2].encode().decode('unicode_escape'))))
>>> rddSplit.first()
(1, 'A', ['Date: 2018/01/01', 'Descr: This is a test des\ncription\n'])
>>> 
>>> #create empty Pandas df
>>> df1 = pd.DataFrame()
>>> 
>>> #insert rows
>>> for rdd in rddSplit.collect():
...     a = {i.split(':')[0].strip():i.split(':')[1].strip('\n').replace('\n','\\n').strip() for i in rdd[2]}
...     a['id'] = rdd[0]
...     a['type'] = rdd[1]
...     df2 = pd.DataFrame([a], columns=a.keys())
...     df1 = pd.concat([df1, df2])
... 
>>> df1
         Date                         Descr                         Warning  id type
0  2018/01/01  This is a test des\ncription                             NaN   1    A
0  2018/01/02     Another test desc\niption                             NaN   2    B
0  2018/01/03                           NaN  This is a warnin\ng, watch out   3    A
>>>
>>> #create spark df
>>> df3 = spark.createDataFrame(df1.fillna('')).replace('',None)
>>> df3.show(truncate=False)
+----------+----------------------------+------------------------------+---+----+
|Date      |Descr                       |Warning                       |id |type|
+----------+----------------------------+------------------------------+---+----+
|2018/01/01|This is a test des\ncription|null                          |1  |A   |
|2018/01/02|Another test desc\niption   |null                          |2  |B   |
|2018/01/03|null                        |This is a warnin\ng, watch out|3  |A   |
+----------+----------------------------+------------------------------+---+----+