如何创建自定义的用户定义聚合不同函数

Question

我有一个包含 4 列的数据帧。

数据帧示例

id1 id2 id3 id4
---------------
a1  a2  a3  a4
b1  b2  b3  b4
b1  b2  b3  b4
c1  c2  c3  c4
    b2      
c1
        a3
            a4
c1
        d4

一行中有 2 种类型的数据，要么所有列都有数据，要么只有一列。

我想对所有列执行不同的功能，例如在比较行之间的值时，它只会比较行中存在的值，而不考虑空值。

输出数据帧应为

id1 id2 id3 id4
a1  a2  a3  a4
b1  b2  b3  b4
c1  c2  c3  c4
        d4

我已经在火花中查看了多个UDAF的例子。但无法根据修改。

Answer 1

您可以使用filter用于所有列，如下所示

df.filter($"id1" =!= "" && $"id2" =!= "" && $"id3" =!= "" && $"id4" =!= "")

您应该获得最终数据帧。

上面的代码适用于静态四列数据帧。如果你有超过四列以上的方法会变得忙碌，因为你将不得不编写太多的逻辑检查。

解决方案是使用如下所示的udf函数

import org.apache.spark.sql.functions._
def checkIfNull = udf((co : mutable.WrappedArray[String]) => !(co.contains(null) || co.contains("")))
df.filter(checkIfNull(array(df.columns.map(col): _*))).show(false)

我希望答案对您有所帮助

Answer 2

可以利用

dropDuplicates依赖于顺序来解决这个问题，请参阅此处的答案。但是，它不是很有效，应该有一个更有效的解决方案。

首先使用 distinct() 删除所有重复项，然后按每列迭代排序并删除其重复项。列按降序排列，nulls然后放在最后。

具有四个静态列的示例：

val df2 = df.distinct()
  .orderBy($"id1".desc).dropDuplicates("id1")
  .orderBy($"id2".desc).dropDuplicates("id2")
  .orderBy($"id3".desc).dropDuplicates("id3")
  .orderBy($"id4".desc).dropDuplicates("id4")