Pyspark 数组列 - 用默认值替换空元素

Question

我有一个数据框，其中有一列是字符串数组。数组的某些元素可能会丢失，如下所示：

-------------|-------------------------------
ID           |array_list                      
---------------------------------------------
38292786     |[AAA,, JLT]                    |
38292787     |[DFG]                          |
38292788     |[SHJ, QKJ, AAA, YTR, CBM]      |
38292789     |[DUY, ANK, QJK, POI, CNM, ADD] |
38292790     |[]                             |
38292791     |[]                             |
38292792     |[,,, HKJ]                      |

我想用默认值“ZZZ”替换缺失的元素。有没有办法做到这一点？我尝试了以下代码，它使用了转换函数和正则表达式：

import pyspark.sql.functions as F
from pyspark.sql.dataframe import DataFrame

def transform(self, f):
    return f(self)

DataFrame.transform = transform  
  
df = df.withColumn("array_list2", F.expr("transform(array_list, x -> regexp_replace(x, '', 'ZZZ'))")) 

这不会产生错误，但会产生废话。我在想我只是不知道识别数组缺失元素的正确方法 - 谁能帮帮我？

在生产中，我们的数据大约有 1000 万行，我试图避免使用 explode 或 UDF（不确定是否可以避免同时使用这两种方法，只需要代码尽可能高效地运行）。我正在使用 Spark 2.4.4

这是我希望输出的样子：

-------------|-------------------------------|-------------------------------
ID           |array_list                     | array_list2
---------------------------------------------|-------------------------------
38292786     |[AAA,, JLT]                    |[AAA, ZZZ, JLT]                     
38292787     |[DFG]                          |[DFG]                          
38292788     |[SHJ, QKJ, AAA, YTR, CBM]      |[SHJ, QKJ, AAA, YTR, CBM]      
38292789     |[DUY, ANK, QJK, POI, CNM, ADD] |[DUY, ANK, QJK, POI, CNM, ADD] 
38292790     |[]                             |[ZZZ]                             
38292791     |[]                             |[ZZZ]   
38292792     |[,,, HKJ]                      |[ZZZ, ZZZ, ZZZ, HKJ]           

Answer 1

regex_replace 在字符级别起作用。

我也无法让它与转换一起使用，但在第一个回答者的帮助下，我使用了 UDF - 没那么容易。

这是我的数据示例，您可以定制。

%python

from pyspark.sql.types import StringType, ArrayType
from pyspark.sql.functions import udf, col

concat_udf = udf(
    lambda con_str, arr: [
        x if x is not None else con_str for x in arr or [None]
    ],
    ArrayType(StringType()),
)

arrayData = [
        ('James',['Java','Scala']),
        ('Michael',['Spark','Java',None]),
        ('Robert',['CSharp','']),
        ('Washington',None),
        ('Jefferson',['1','2'])]

df = spark.createDataFrame(data=arrayData, schema = ['name','knownLanguages'])
df = df.withColumn("knownLanguages", concat_udf(lit("ZZZ"), col("knownLanguages")))
df.show()

返回：

+----------+------------------+
|      name|    knownLanguages|
+----------+------------------+
|     James|     [Java, Scala]|
|   Michael|[Spark, Java, ZZZ]|
|    Robert|        [CSharp, ]|
|Washington|             [ZZZ]|
| Jefferson|            [1, 2]|
+----------+------------------+

这很困难，得到了第一个回答者的帮助。

Answer 2

我正在考虑一些事情，但我不确定它是否有效。

from pyspark.sql import functions as F

df.withColumn("array_list2", F.split(F.array_join("array_list", ",", "ZZZ"), ","))

首先，我将值连接为带有分隔符, 的字符串（希望您的字符串中没有它，但您可以使用其他东西）。我使用null_replacement 选项来填充null 值。然后我按照相同的分隔符进行拆分。

---

编辑：根据@thebluephantom 评论，您可以试试这个解决方案：

df.withColumn(
    "array_list_2", F.expr(" transform(array_list, x -> coalesce(x, 'ZZZ'))")
).show()

SQL 内置 transform 不适合我，所以我无法尝试，但希望你会得到你想要的结果。