Spark中如何处理结构体数组字段生成subscriberPresent布尔列？

现有如下Spark DataFrame：

数据结构 #

df.printSchema()
root
 |-- code: string (nullable = true)
 |-- contractId: string (nullable = true)
 |-- contractArray: array (nullable = false)
 |    |-- element: struct (containsNull = false)
 |    |    |-- profile: string (nullable = true)
 |    |    |-- id: string (nullable = true)

数据内容 #

df.show()
+-----+----------+----------------------------------------+
| code|contractId|                           contractArray|
+-----+----------+----------------------------------------+
|    A|     45 8 |    [{CONSUMER, 789}, {SUBSCRIBER, 789}]|
|   AC|   7896 0 |                      [{CONSUMER, null}]|
|   BB|     12 7 |  [{CONSUMER, null}, {SUBSCRIBER, null}]|
|  CCC|    753 8 |  [{SUBSCRIBER, null}, {CONSUMER, 7854}]|
+-----+----------+----------------------------------------+

需求 #

需要新增布尔列subscriberPresent，判断规则：当contractArray中存在profile为"SUBSCRIBER"且id不为null的元素时，值为true，否则为false。

期望结果 #

+-----+----------+----------------------------------------+-----------------+
| code|contractId|                           contractArray|subscriberPresent|
+-----+----------+----------------------------------------+-----------------+
|    A|     45 8 |    [{CONSUMER, 789}, {SUBSCRIBER, 789}]|             true|
|   AC|   7896 0 |                      [{CONSUMER, null}]|            false|
|   BB|     12 7 |  [{CONSUMER, null}, {SUBSCRIBER, null}]|            false|
|  CCC|    753 8 |  [{SUBSCRIBER, null}, {CONSUMER, 7854}]|            false|
+-----+----------+----------------------------------------+-----------------+

原本打算用UDF实现，想了解更优的替代方案。

方案1：使用Spark内置exists函数（Spark 3.0+推荐） #

Spark 3.0及以上版本支持exists高阶函数，可直接对数组进行条件判断，无需展开数组，性能最优。

代码示例：

import org.apache.spark.sql.functions.{exists, col}

val resultDF = df.withColumn(
  "subscriberPresent",
  exists(
    col("contractArray"),
    elem => elem("profile") === "SUBSCRIBER" && elem("id").isNotNull
  )
)

方案2：使用filter+size函数（兼容Spark 2.x） #

如果使用Spark 2.x版本，无exists函数支持，可通过filter筛选符合条件的数组元素，再判断筛选后数组的长度是否大于0。

代码示例：

import org.apache.spark.sql.functions.{filter, size, col}

val resultDF = df.withColumn(
  "subscriberPresent",
  size(
    filter(
      col("contractArray"),
      elem => elem("profile") === "SUBSCRIBER" && elem("id").isNotNull
    )
  ) > 0
)

不推荐使用UDF的原因 #

UDF虽然能实现需求，但存在明显劣势：

• 性能差：无法被Spark Catalyst优化器优化，需要额外的序列化/反序列化操作，运行效率远低于内置函数。
• 类型风险：需手动处理数据类型转换，容易引发运行时错误。
• 维护成本高：内置函数是Spark原生支持的标准API，代码更简洁易懂，无需额外维护自定义函数逻辑。

内容的提问来源于stack exchange，提问作者Mamaf