使用命名的regex从PySpark数据帧中的列中提取多个列

Question

假设我在pySpark中有一个数据帧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 |

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

我使用命名组创建了一个regex语句，可用于从description字段提取相关信息，大致如下：

^{pr2}$

同样，伪正则表达式，实际的正则表达式稍微复杂一些，但目的是捕获三个可能的组：

| 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 |

现在我想将regex匹配结果的列添加到原始数据帧中（即，将这个问题中的两个伪表合并为一个）。在

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

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()，但是在Row()-构造函数中所需的列顺序（以及我无法硬编码将由正则表达式组添加的列名）的问题，从而导致数据帧中的列全部混乱。如何实现我想要的，即一个包含六列的数据帧：id，type，description，DATE，DESCR和{}？在

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

Answer 1

我想你可以在这个案子中使用熊猫的特征。首先，我将df转换为rdd来分割描述字段。我拉一个熊猫df然后我用熊猫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   |
+     +              +               + -+  +