在Databricks Workbook中用Spark SQL实现SQL UDF间参数传递

问题结论 #

Databricks SQL目前不支持在SQL表值UDF（TVF）的定义内部直接调用另一个SQL表值UDF，这就是你执行tbl_func时触发AnalysisException: could not resolve 'tbl_filter' to a table-valued function错误的原因。

替代解决方案 #

针对你希望复用过滤逻辑、避免代码重复的需求，推荐以下几种可行方案：

1. 用CTE/内联子查询复用过滤逻辑

将tbl_filter的过滤逻辑封装为CTE，在各个tbl_func版本中直接引用，既保证逻辑统一，又无需重复编写：

CREATE OR REPLACE TEMPORARY VIEW test_tbl AS
WITH a AS
(SELECT explode(sequence(1, 10)) AS a),
b AS
(SELECT explode(sequence(50, 60)) AS b)
SELECT * FROM a CROSS JOIN b;

CREATE OR REPLACE TEMPORARY FUNCTION tbl_func(a_val INT, b_val INT)
    RETURNS TABLE(a INT, b INT)
    RETURN
    WITH filtered_by_a AS (
        SELECT * FROM test_tbl WHERE a = a_val
    )
    SELECT * FROM filtered_by_a WHERE b = b_val;

-- 调用验证
SELECT * FROM tbl_func(1, 60);

2. 标量UDF配合参数化视图

如果过滤逻辑可以拆分为布尔判断，先定义标量UDF实现核心逻辑，再通过参数化视图复用：

-- 定义标量UDF实现a值过滤逻辑
CREATE OR REPLACE TEMPORARY FUNCTION is_match_a(a_col INT, target_a INT)
RETURNS BOOLEAN
RETURN a_col = target_a;

-- 创建参数化视图，基于标量UDF过滤数据
CREATE OR REPLACE TEMPORARY VIEW filtered_by_a(target_a) AS
SELECT * FROM test_tbl WHERE is_match_a(a, target_a);

-- 在tbl_func中调用参数化视图
CREATE OR REPLACE TEMPORARY FUNCTION tbl_func(a_val INT, b_val INT)
    RETURNS TABLE(a INT, b INT)
    RETURN
    SELECT * FROM filtered_by_a(a_val) WHERE b = b_val;

-- 调用验证
SELECT * FROM tbl_func(1, 60);

3. 改用编程语言实现UDTF（推荐复杂场景）

如果SQL UDF的限制无法满足需求，可以用Python/Scala编写用户定义表值函数（UDTF），这类函数支持内部调用其他函数，完美实现逻辑复用：

以Python UDTF为例：

from pyspark.sql.types import StructType, StructField, IntegerType

# 实现tbl_filter的UDTF
class TblFilter:
    def eval(self, a_val):
        # 直接查询并返回符合条件的行
        df = spark.sql(f"SELECT a, b FROM test_tbl WHERE a = {a_val}")
        for row in df.collect():
            yield (row.a, row.b)

# 注册UDTF
spark.udf.register("tbl_filter_udtf", TblFilter, StructType([
    StructField("a", IntegerType()),
    StructField("b", IntegerType())
]))

# 实现依赖tbl_filter的tbl_func UDTF
class TblFunc:
    def eval(self, a_val, b_val):
        # 调用已注册的tbl_filter UDTF
        filtered_df = spark.sql(f"SELECT a, b FROM tbl_filter_udtf({a_val})")
        for row in filtered_df.collect():
            if row.b == b_val:
                yield (row.a, row.b)

# 注册tbl_func UDTF
spark.udf.register("tbl_func_udtf", TblFunc, StructType([
    StructField("a", IntegerType()),
    StructField("b", IntegerType())
]))

调用验证：

SELECT * FROM tbl_func_udtf(1, 60);

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