计算pyspark df列中子字符串列表的出现次数

Question

我想计算子字符串列表的出现次数，并基于pyspark df中包含一个长字符串的列创建一个列。在

Input:          
       ID    History

       1     USA|UK|IND|DEN|MAL|SWE|AUS
       2     USA|UK|PAK|NOR
       3     NOR|NZE
       4     IND|PAK|NOR

 lst=['USA','IND','DEN']


Output :
       ID    History                      Count

       1     USA|UK|IND|DEN|MAL|SWE|AUS    3
       2     USA|UK|PAK|NOR                1
       3     NOR|NZE                       0
       4     IND|PAK|NOR                   1

Answer 1

# Importing requisite packages and creating a DataFrame
from pyspark.sql.functions import split, col, size, regexp_replace
values = [(1,'USA|UK|IND|DEN|MAL|SWE|AUS'),(2,'USA|UK|PAK|NOR'),(3,'NOR|NZE'),(4,'IND|PAK|NOR')]
df = sqlContext.createDataFrame(values,['ID','History'])
df.show(truncate=False)
+ -+             +
|ID |History                   |
+ -+             +
|1  |USA|UK|IND|DEN|MAL|SWE|AUS|
|2  |USA|UK|PAK|NOR            |
|3  |NOR|NZE                   |
|4  |IND|PAK|NOR               |
+ -+             +

其思想是根据这三个delimiters:lst=['USA','IND','DEN']拆分字符串，然后计算生成的子字符串的数量。在

例如，字符串USA|UK|IND|DEN|MAL|SWE|AUS被拆分为-,，|UK|，|，|MAL|SWE|AUS。因为创建了4个子字符串，并且有3个分隔符匹配，所以4-1 = 3给出了出现在列字符串中的这些字符串的计数。在

我不确定Spark是否支持多字符分隔符，因此作为第一步，我们将列表['USA','IND','DEN']中的这3个子字符串中的任何一个替换为一个标志/伪值%。你也可以用别的东西。以下代码执行此操作^{}-

^{pr2}$

最后，我们计算由^{}创建的子字符串的数量，首先使用%作为分隔符，然后计算使用^{}函数创建的子字符串的数量，最后从中减去1。在

df = df.withColumn('Count', size(split(col('History_X'), "%")) - 1).drop('History_X')
df.show(truncate=False)
+ -+             +  -+
|ID |History                   |Count|
+ -+             +  -+
|1  |USA|UK|IND|DEN|MAL|SWE|AUS|3    |
|2  |USA|UK|PAK|NOR            |1    |
|3  |NOR|NZE                   |0    |
|4  |IND|PAK|NOR               |1    |
+ -+             +  -+

Answer 2

如果您使用的是Spark 2.4+，可以尝试Spark SQL高阶函数^{}：

from pyspark.sql import functions as F

>>> df.show(5,0)
+ -+             +
|ID |History                   |
+ -+             +
|1  |USA|UK|IND|DEN|MAL|SWE|AUS|
|2  |USA|UK|PAK|NOR            |
|3  |NOR|NZE                   |
|4  |IND|PAK|NOR               |
+ -+             +

df_new = df.withColumn('data', F.split('History', '\|')) \
           .withColumn('cnt', F.expr('size(filter(data, x -> x in ("USA", "IND", "DEN")))'))

>>> df_new.show(5,0)
+ -+             +                 + -+
|ID |History                   |data                              |cnt|
+ -+             +                 + -+
|1  |USA|UK|IND|DEN|MAL|SWE|AUS|[USA, UK, IND, DEN, MAL, SWE, AUS]|3  |
|2  |USA|UK|PAK|NOR            |[USA, UK, PAK, NOR]               |1  |
|3  |NOR|NZE                   |[NOR, NZE]                        |0  |
|4  |IND|PAK|NOR               |[IND, PAK, NOR]                   |1  |
+ -+             +                 + -+

其中我们首先将字段History拆分为一个名为data的数组列，然后使用filter函数：

^{pr2}$

为了只检索满足条件的数组元素：IN ("USA", "IND", "DEN")，之后，我们使用size()函数对结果数组进行计数。在

更新：添加了另一种使用array_contains()的方法，该方法应适用于旧版本Spark：

lst = ["USA", "IND", "DEN"]

df_new = df.withColumn('data', F.split('History', '\|')) \
           .withColumn('Count', sum([F.when(F.array_contains('data',e),1).otherwise(0) for e in lst]))

注意：将跳过数组中的重复项，此方法只计算唯一的国家/地区代码。在