apache-spark - pyspark：返回所有单元格匹配正则表达式的列

Question

我有一个包含多列的 spark 数据框，每列包含一个字符串。

例如输入： +--------------+--------------+--------------+--------------+ | c1| c2 | c3| c4| +--------------+--------------+--------------+--------------+ |11 - 12 - 1993| 4 | 4 | 2014 | 8 - 7 - 2013 | null | |12 / 6 / 1965 | 8 - 6 - 2013 | date missing |11 - 12 - 1993| |10 / 5 / 2001 | 7 - 11 - 2011| 4 | 5 | 2015 | 10 / 5 / 2001| +--------------+--------------+--------------+--------------+

我需要返回所有值与特定正则表达式模式匹配的列。

在这个例子中，我必须返回所有值都是有效日期的列。也就是说，在这种情况下，应返回 C1 和 C2 列中的所有值都有一个有效日期（无论其格式如何）。

例如输出 +--------------+--------------+ | c1| c2 | +--------------+--------------+ |11 - 12 - 1993| 4 | 4 | 2014 | |12 / 6 / 1965 | 8 - 6 - 2013 | |10 / 5 / 2001 | 7 - 11 - 2011| +--------------+--------------+

我有正则表达式。最好的方法是什么？我想找出最有效的方法来做到这一点。

Answer 1

一种不使用正则表达式的方法：

from pyspark.sql.functions import coalesce, to_date

# list of all expected date format
fmts = ['MM - d - yyyy', 'MM / d / yyyy', 'MM | d | yyyy']

# columns to check
cols = df.columns

# convert columns into date using coalesce and to_date on all available fmts 
# convert the resulting column to StringType (as df.summary() doesn't work on DateType)
df1 = df.select([ coalesce(*[to_date(c, format=f) for f in fmts]).astype('string').alias(c) for c in cols])
df1.show()
+----------+----------+----------+----------+
|        c1|        c2|        c3|        c4|
+----------+----------+----------+----------+
|1993-11-12|2014-04-04|2013-08-07|      null|
|1965-12-06|2013-08-06|      null|1993-11-12|
|2001-10-05|2011-07-11|2015-04-05|2001-10-05|
+----------+----------+----------+----------+

现在，如果列包含任何空值，您的任务将变为计数。

# get all Number of rows in df
N = df.count()
# 3

# use df.summary('count') find all non-null #Row for each columns
df1.summary('count').show()                                                                                        
+-------+---+---+---+---+
|summary| c1| c2| c3| c4|
+-------+---+---+---+---+
|  count|  3|  3|  2|  2|
+-------+---+---+---+---+

找到具有以下内容的列名count == N：

cols_keep = [ c for c,v in df1.summary('count').select(cols).first().asDict().items() if int(v) == N ]
# ['c1', 'c2']

df_new = df.select(cols_keep)

使用正则表达式

如果你想使用你的正则表达式来处理这个任务：

from pyspark.sql.functions import regexp_extract

# pattern should be very complex in order to effectively validate dates. this one is just for testing
ptn = r'\d+ [-/|] \d+ [-/|] \d+'

df1 = df.select([ regexp_extract(c, ptn, 0).alias(c) for c in cols ] ).replace('', None)
+--------------+-------------+------------+--------------+
|            c1|           c2|          c3|            c4|
+--------------+-------------+------------+--------------+
|11 - 12 - 1993| 4 | 4 | 2014|8 - 7 - 2013|          null|
| 12 / 6 / 1965| 8 - 6 - 2013|        null|11 - 12 - 1993|
| 10 / 5 / 2001|7 - 11 - 2011|4 | 5 | 2015| 10 / 5 / 2001|
+--------------+-------------+------------+--------------+

那么这变得与上面相同，以查找和排除包含空值的列。

笔记：

您可以使用以下方式一次性完成此操作df.summary()：

from pyspark.sql.functions import regexp_extract, sum, when 

df1 = df.select([ sum(when(regexp_extract(c, ptn, 0)== '',1).otherwise(0)).alias(c) for c in cols ] )
+---+---+---+---+
| c1| c2| c3| c4|
+---+---+---+---+
|  0|  0|  1|  1|
+---+---+---+---+

cols_keep = [ c for c,v in df1.first().asDict().items() if not v ]

类似于以前使用正则表达式的方法。