Spark UDF在测试DataFrame生效但原始DataFrame返回全null问题排查

我有一个存储了hive_metastore中Delta表列表的DataFrame，想通过UDF获取每张表的Delta Log路径来提取信息。原本可以把DataFrame收集成数组后逐行处理，但为了提升效率尝试用UDF实现。我写的udfGetDeltaLog UDF如下：

import org.apache.spark.sql.functions.udf

val udfGetDeltaLog = udf(
  (catalog: String, database: String, table: String) => {
    val deltaLog = try {
      Some(DeltaLog.forTable(spark, TableIdentifier(table, Some(database), Some(catalog))))
    } catch {
      case e: Throwable => None
    }

    deltaLog match {
      case Some(log) => log.logPath.toString
      case None => null
    }
  }
)

spark.udf.register("udfGetDeltaLog", udfGetDeltaLog)

原始DataFrame的结构及部分数据：

inputTables.printSchema

root
 |-- catalog_name: string (nullable = true)
 |-- database_name: string (nullable = true)
 |-- table_name: string (nullable = true)

inputTables.limit(3).show(false)

+--------------+-------------+-------------------------------------------+
|catalog_name  |database_name|table_name                                 |
+--------------+-------------+-------------------------------------------+
|hive_metastore|adw_poc      |table1                                     |
|hive_metastore|alfam        |table2                                     |
|hive_metastore|alfam        |table3                                     |
+--------------+-------------+-------------------------------------------+

我手动构造了结构一致的测试DataFrame：

val df = Seq(
  ("hive_metastore", "adw_poc", "table1"), 
  ("hive_metastore", "alfam", "table2"), 
  ("hive_metastore", "alfam", "table3")
).toDF("catalog_name", "database_name", "table_name")

df.printSchema

root
 |-- catalog_name: string (nullable = true)
 |-- database_name: string (nullable = true)
 |-- table_name: string (nullable = true)

UDF在测试DataFrame上能正常返回Delta Log路径：

df
  .withColumn("dl", udfGetDeltaLog($"catalog_name", $"database_name", $"table_name"))
  .show(false)

+-------------------------------------------------------------+
|dl                                                           |
+-------------------------------------------------------------+
|dbfs:/user/hive/warehouse/adw_poc.db/table1/_delta_log       |
|dbfs:/user/hive/warehouse/alfam.db/table2/_delta_log         |
|dbfs:/user/hive/warehouse/alfam.db/table3/_delta_log         |
+-------------------------------------------------------------+

但应用到原始DataFrame时却全返回null：

+----+
|dl  |
+----+
|null|
|null|
|null|
+----+

请问这种差异的原因是什么？

出现这种差异的核心原因集中在分布式执行环境的序列化问题、原始数据的隐性异常或权限差异，具体拆解如下：

1. Spark上下文的序列化问题 #

你的UDF直接引用了外部的spark变量，在Spark分布式执行时，这个变量需要序列化后发送到各个Executor节点。虽然SparkSession理论上可序列化，但在集群环境的类加载差异、会话实例传递限制等场景下，Executor端可能无法正常初始化会话，导致调用DeltaLog.forTable时抛出异常，最终返回null。

而测试DataFrame是Driver端本地生成的小数据量，触发了Spark本地执行模式，不需要序列化SparkSession，因此能正常运行。

修复方案：
改用SparkSession.getActiveSession在UDF内部获取当前会话，避免外部变量序列化问题：

val udfGetDeltaLog = udf(
  (catalog: String, database: String, table: String) => {
    val deltaLog = try {
      SparkSession.getActiveSession.flatMap { spark =>
        Some(DeltaLog.forTable(spark, TableIdentifier(table, Some(database), Some(catalog))))
      }
    } catch {
      case e: Throwable => None
    }

    deltaLog match {
      case Some(log) => log.logPath.toString
      case None => null
    }
  }
)

2. 原始DataFrame字段的隐性异常 #

虽然原始DataFrame的Schema显示字段为string类型，但实际数据可能存在：

• 字段值包含不可见字符（如空格、制表符），导致TableIdentifier无法匹配真实表
• 部分字段为隐性null（即使show输出看起来有值），比如catalog_name或database_name存在空值
• 表名大小写不匹配：Hive metastore中的表名可能大小写敏感，测试数据是精准匹配，而原始DataFrame的表名存在大小写差异

排查与修复：
先对原始DataFrame做清洗，去除字段首尾空格并过滤空值：

val cleanedInputTables = inputTables
  .withColumn("catalog_name", trim($"catalog_name"))
  .withColumn("database_name", trim($"database_name"))
  .withColumn("table_name", trim($"table_name"))
  .filter($"catalog_name".isNotNull && $"database_name".isNotNull && $"table_name".isNotNull)

// 再应用UDF
cleanedInputTables.withColumn("dl", udfGetDeltaLog($"catalog_name", $"database_name", $"table_name"))

3. Executor端的权限差异 #

测试DataFrame在Driver端执行时，使用的是Driver进程的权限，能正常访问Delta表的元数据和存储路径。但集群环境中，Executor节点的进程权限可能没有访问对应DBFS路径的权限，导致DeltaLog.forTable调用失败，返回null。

排查方法：
检查Executor节点用户是否有访问目标Delta表存储路径（如dbfs:/user/hive/warehouse/adw_poc.db/table1/）的权限，或在集群环境中运行测试代码，验证是否同样返回null，以此确认是否为权限问题。

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