UDF 连接埋在行对象中的未定义大小写类的数组

Question

我有一个数据框架，叫做sessions，它的列可能会随着时间而变化。(编辑以澄清:我没有列的case类——只有反射模式。)我将在外部作用域中始终有一个uuid和clientId，其他一些内部和外部作用域列可能会构成跟踪事件，所以...类似于:

root
 |-- runtimestamp: long (nullable = true)
 |-- clientId: long (nullable = true)
 |-- uuid: string (nullable = true)
 |-- oldTrackingEvents: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- timestamp: long (nullable = true)
 |    |    |-- actionid: integer (nullable = true)
 |    |    |-- actiontype: string (nullable = true)
 |    |    |-- <tbd ... maps, arrays and other stuff matches sibling> section
 ...
 |-- newTrackingEvents: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- timestamp: long (nullable = true)
 |    |    |-- actionid: integer (nullable = true)
 |    |    |-- actiontype: string (nullable = true)
 |    |    |-- <tbd ... maps, arrays and other stuff matches sibling>      
 ...

我现在想将oldTrackingEvents和newTrackingEvents与包含这些参数和尚待解决的代码逻辑的UDF合并：

val mergeTEs = udf((oldTEs : Seq[Row], newTEs :  Seq[Row]) =>
        // do some stuff - figure best way 
        //    - to merge both groups of tracking events
        //    - remove duplicates tracker events structures
        //    - limit total tracking events < 500 
        return result // same type as UDF input params
    )

UDF返回结果将是结构的数组，它是连接的两个字段的结果List。

问题：我的问题是如何构造这样的UDF-（1）使用正确的传入参数类型，（2）在UDF中操作这些集合的方法，（3）返回没有编译器错误的值的明确方法。我未成功测试输入/输出的Seq[Row]（使用val testUDF=udf（（跟踪事件：Seq[Row]））=

sessions.select(mergeTEs('oldTrackingEvents, 'newTrackingEvents).as("cleanTrackingEvents"))

在每一行中，...以内存/速度有效的方式取回单个“跟踪事件”结构数组。

补足的：

看看向我展示的一个问题…如果存在相关性，可能会有一个提示…定义一个UDF，接受Spark数据帧中的对象数组<代码>创建传递给udf的结构函数必须返回产品类型（元组*或案例类），而不是行 也许…这篇文章是相关/有用的。




Answer 1

或者。。。如果您的目标是将某些随机行结构/模式的序列与某些操作合并，这是一种避免分区讨论的替代通用方法：

从主数据帧中，为每个trackingEvents部分创建数据帧，< code>new和< code>old。选择每个展开的“跟踪事件”部分的列。将这些< code>val dataframe声明保存为< code>newTE和< code>oldTE。

创建另一个dataframe，其中选取的列对于< code>oldTrackingEvents和< code>newTrackingEvents数组中的每个跟踪事件是唯一的，例如每个事件的< code>uuid 、< code>clientId和事件< code >时间戳。您的伪模式应该是:

< code>(uuid: String，clientId : Long，newTE : Seq[Long]，oldTE : Seq[Long])

使用UDF连接结构中的两个简单序列，两者都是< code>Seq[Long]，这是“未经测试的东西”示例:

val limitEventsUDF = udf { (newTE: Seq[Long], oldTE: Seq[Long], limit: Int, tooOld: Long) => {
    (newTE ++ oldTE).filter(_ > tooOld).sortWith(_ > _).distinct.take(limit) 
}}

UDF将返回清除跟踪事件的数据帧

之后根据需要使用collect_ list进行分组。

仍然...这看起来工作量很大，我不确定是否应该投票给< code >“答案”？

Answer 2

匿名用户

我认为你所链接的问题解释了这一切，所以我重申一下。使用自定义项时：

• STRtType的输入表示形式是弱类型的Row对象。
• 结构类型的输出类型必须是Scala产品。您不能返回Row对象。

如果这会带来很大的负担，则应使用强类型数据集

val f: T => U 
sessions.as[T].map(f): Dataset[U]

其中< code>T是表示< code >会话模式的代数数据类型，而< code>U是表示结果的代数数据类型。