如何将结构或类的数组从UDF返回到数据框列值？ [英] How to return an array of struct or class from UDF into dataframe column value?

问题描述

d = [{'ID': '1', 'pID': 1000, 'startTime':'2018.07.02T03:34:20', 'endTime':'2018.07.03T02:40:20'}, {'ID': '1', 'pID': 1000, 'startTime':'2018.07.02T03:45:20', 'endTime':'2018.07.03T02:50:20'}, {'ID': '2', 'pID': 2000, 'startTime':'2018.07.02T03:34:20', 'endTime':'2018.07.03T02:40:20'}, {'ID': '2', 'pID': 2000, 'startTime':'2018.07.02T03:45:20', 'endTime':'2018.07.03T02:50:20'}]

df = spark.createDataFrame(d)

Dates = namedtuple("Dates", "startTime endTime")


def MergeAdjacentUsage(timeSets):
  DatesArray = []
  for times in timeSets:
    DatesArray.append(Dates(startTime=times.startTime, endTime=times.endTime))
  return DatesArray


MergeAdjacentUsages = udf(MergeAdjacentUsage,ArrayType(Dates()))

df1=df.groupBy(['ID','pID']).agg(MergeAdjacentUsages(F.collect_list(struct('startTime','endTime'))).alias("Times"))

display(df1)

我想要的只是设置colum UDF返回的结构数组的n值。它给我的错误是：

All I want is to set column value to an array of stuct that is returned by UDF. It is giving me error as:

TypeError： new （）恰好接受3个参数（给定1个）

TypeError: new() takes exactly 3 arguments (1 given)

TypeError Traceback（最近一次调用
最后）在（）
22返回DatesArray
23
---> 24 MergeAdjacentUsages = udf（MergeAdjacentUsage，ArrayType（Dates（）））
25
26 df1 = df.groupBy（['ID'，'pID']）。agg（MergeAdjacentUsages（F.collect_list（struct（ 'startTime'，'endTime'）））。alias（ Times））

TypeError Traceback (most recent call last) in () 22 return DatesArray 23 ---> 24 MergeAdjacentUsages = udf(MergeAdjacentUsage,ArrayType(Dates())) 25 26 df1=df.groupBy(['ID','pID']).agg(MergeAdjacentUsages(F.collect_list(struct('startTime','endTime'))).alias("Times"))

任何帮助，想法或提示将不胜感激

Any help, idea or hint will be appreciated.

推荐答案

pyspark不允许用户定义的Class对象作为Dataframe列类型。相反，我们需要创建 StructType ，其用法类似于python中的类/命名元组。

pyspark does not let user defined Class objects as Dataframe Column Types. Instead we need to create the StructType which can be used similar to a class / named tuple in python.

例如：

from pyspark.sql.types import *
from pyspark.sql.functions import udf
from pyspark.sql import functions as F
# from pyspark.sql.functions import *

d = [{'ID': '1', 'pID': 1000, 'startTime': '2018.07.02T03:34:20', 'endTime': '2018.07.03T02:40:20'},
     {'ID': '1', 'pID': 1000, 'startTime': '2018.07.02T03:45:20', 'endTime': '2018.07.03T02:50:20'},
     {'ID': '2', 'pID': 2000, 'startTime': '2018.07.02T03:34:20', 'endTime': '2018.07.03T02:40:20'},
     {'ID': '2', 'pID': 2000, 'startTime': '2018.07.02T03:45:20', 'endTime': '2018.07.03T02:50:20'}]

df = spark.createDataFrame(d)

# Dates = namedtuple("Dates", "startTime endTime")

schema = ArrayType(StructType([
    StructField("startTime", StringType(), False),
    StructField("endTime", StringType(), False)
]))


MergeAdjacentUsages = udf(lambda xs: xs, schema)

df1 = df.groupBy(['ID', 'pID']).agg(MergeAdjacentUsages(
    F.collect_list(F.struct('startTime', 'endTime'))).alias("Times"))
df1.show(truncate=False)

+---+----+----------------------------------------------------------------------------------------+
|ID |pID |Times                                                                                   |
+---+----+----------------------------------------------------------------------------------------+
|2  |2000|[[2018.07.02T03:34:20, 2018.07.03T02:40:20], [2018.07.02T03:45:20, 2018.07.03T02:50:20]]|
|1  |1000|[[2018.07.02T03:34:20, 2018.07.03T02:40:20], [2018.07.02T03:45:20, 2018.07.03T02:50:20]]|
+---+----+----------------------------------------------------------------------------------------+

希望这会有所帮助！

这篇关于如何将结构或类的数组从UDF返回到数据框列值？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！