对pyspark中数组列的数组中的项目进行排序和迭代

Question

在 pyspark 中，我有一个由两列组成的数据框

假设数组数组中的细节是 时间戳、电子邮件、电话号码、名字、姓氏、地址、城市、国家/地区、randomId

+-------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| str1                    | array_of_str                                                                                                                                                                                                                                                                                                                                                                                |
+-------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| random column data1     | [[‘2020-01-26 17:30:57.000 +0000’, ’’, ‘728-802-5766’, ‘’, ‘’, ‘7th street crossroads’, ‘seattle’, ‘’, ‘randomId104’], [‘2019-07-20 20:54:57.000 +0000’, ’abaudi0@patch.com’, ‘728-802-5766’, ‘Katuscha’, ‘’, ‘’, ‘’, ‘us’, ‘randomId225’], [‘2015-12-04 04:54:57.000 +0000’, ’abaudi0-temp@patch.com’, ‘728-802-5766’, ‘’, ‘Othen’, ‘7th street crossroads’, ‘seattle’, ‘’, ‘randomid306’]]|
| random column data2     | [[‘2021-01-30 17:30:04.000 +0000’, ’jallmann2@digg.com’, ‘313-984-9692’, ‘’, ‘’, ‘th street crossroads’, ‘New york’, ‘us’, ‘randomId563’], [‘2018-05-15 20:44:57.000 +0000’, ’cgodlee6@sphinn.com’, ‘’, ‘Marianne’, ‘Allmann’, ‘’, ‘’, ‘us’, ‘randomId884’]]                                                                                                                                |
+-------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

我期待像下面这样的输出数据框

+-------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------+
| str1                    | array_of_str                                                                                                                                         |
+-------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------+
| random column data1     | [‘2020-01-26 17:30:57.000 +0000’, ’abaudi0@patch.com’, ‘728-802-5766’, ‘Katuscha’, ‘Othen’, ‘7th street crossroads’, ‘seattle’, ‘us’, ‘randomid306’] |
| random column data2     | [‘2021-01-30 17:30:04.000 +0000’, ’jallmann2@digg.com’, ‘313-984-9692’, ‘Marianne’, ‘Allmann’, ‘111th Ave NE’, ‘New york’, ‘us’, ‘randomId884’]      |
+-------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------- +

可选：- 数组数组中的现有数据可能尚未按时间戳递减顺序排序。如何按时间戳递减顺序对数组数组进行排序

在这里，我打算编写一个 udf 来从数组数组中提取最新的非空（时间戳、电子邮件、电话号码、名字、姓氏、地址、城市、国家/地区）数据。如果是randomId，我会一直拉取系统中最旧记录的randomId。

示例：- 对于random column data1 emailId，即abaudi0@patch.com 是从数组中的第二个元素填充的，因为第一个元素的电子邮件 ID 为空。 其他列的情况类似。 如果randomid randomid306 第一条记录是最旧的条目，因此它填充在我的输出数据框中。

在udf中

1. 如何按时间戳降序对数组元素的数组进行排序？ - 一种可选步骤

2. 如何遍历数据框中的数组列？

3)如何访问 udf 中数组上的单个项目？

就像在 python 的情况下，我们可以迭代列表元素列表，如

for item in items:
   print(item[0], item[1])

如何为 pyspark 中的数组列实现类似的功能？

我可以通过不将数据转换为 pandas 数据帧来在 pyspark 中执行上述步骤吗？

火花版本 2.4.3 蟒蛇3.6.8

Answer 1

您可以为此使用内置的sort_array，使用它创建一个新列并使用getItem 提取第一个元素(0)

内置sort_array

input_list = [('10',[
                  ['2020-01-26 17:30:57.000 +0000', '', '728-802-5766', '', '', '7th street crossroads', 'seattle', '', 'randomId104']
                , ['2019-07-20 20:54:57.000 +0000', 'abaudi0@patch.com', '728-802-5766', 'Katuscha', '', '', '', 'us', 'randomId225']
                , ['2015-12-04 04:54:57.000 +0000', 'abaudi0-temp@patch.com', '728-802-5766', '', 'Othen', '7th street crossroads', 'seattle', '', 'randomid306']
            ])
        ,('20',[
          ['2021-01-30 17:30:04.000 +0000', 'jallmann2@digg.com', '313-984-9692', '', '', 'th street crossroads', 'New york', 'us', 'randomId563']
        , ['2018-05-15 20:44:57.000 +0000', 'cgodlee6@sphinn.com', '', 'Marianne', 'Allmann', '', '', 'us', 'randomId884']
        ])
    ]

sparkDF = sql.createDataFrame(input_list,['id','array_str'])


sparkDF = sparkDF.withColumn('sorted_array_str'
                                 ,F.sort_array(F.col('array_str'),False).getItem(0))


sparkDF.select(['id','sorted_array_str']).show(truncate=False)

UDF

# array_sort_udf, sorts on the timestamp , can be futher customized for giving precedence to multiple elements

array_sort_udf = F.udf(lambda x : sorted(x,key=lambda x: x[0], reverse=True), ArrayType(StringType()))

sparkDF = sparkDF.withColumn('sorted_array_str'
                                 ,array_sort_udf(F.col('array_str')).getItem(0))

sparkDF.select(['id','sorted_array_str']).show(truncate=False)

输出

+---+----------------------------------------------------------------------------------------------------------------------+
|id |sorted_array_str                                                                                                      |
+---+----------------------------------------------------------------------------------------------------------------------+
|10 |[2020-01-26 17:30:57.000 +0000, , 728-802-5766, , , 7th street crossroads, seattle, , randomId104]                    |
|20 |[2021-01-30 17:30:04.000 +0000, jallmann2@digg.com, 313-984-9692, , , th street crossroads, New york, us, randomId563]|
+---+----------------------------------------------------------------------------------------------------------------------+