Spark SQL在UDF中使用Join与子查询报错问题求助

你错误地将全局DataFrame转换逻辑封装成了UDF。UDF是用于行级数据处理的，运行在Worker节点，而spark.sql属于Driver端的SparkSession操作，无法序列化到Worker执行——哪怕你简化SQL只做SELECT，只要函数里调用了spark.sql，就会在Worker端尝试访问SparkContext，必然触发这个权限限制错误。

1. 放弃注册UDF，直接以普通函数调用 #

你的breadth_first_search本质是对整个DataFrame的批量转换操作，不需要注册成UDF，直接作为普通函数传入DataFrame即可。

2. 修正SQL中的语法错误 #

原SQL存在字段名/别名不匹配、表别名使用错误的问题，修正后如下：

def breadth_first_search(previously_traversed_nodes):
    # 将输入DataFrame注册为临时视图，让SQL可以访问
    previously_traversed_nodes.createOrReplaceTempView("previously_traversed_nodes")
    
    sql_query = '''
                SELECT person, separation
                FROM previously_traversed_nodes

                UNION

                SELECT
                  pg.end_node AS person, ptn.separation + 1 AS separation
                FROM
                  previously_traversed_nodes AS ptn
                JOIN
                  person_graph pg
                ON ptn.person = pg.start_node
                WHERE
                  pg.end_node NOT IN (
                    SELECT DISTINCT(person) FROM previously_traversed_nodes
                  )
           '''

    result = spark.sql(sql_query)
    return result

注意：需确保person_graph已经注册为Spark可访问的临时视图或外部表。

3. 正确调用方式 #

直接传入初始遍历节点的DataFrame即可，无需注册UDF：

# 初始化起始节点DataFrame
initial_nodes = spark.createDataFrame([("Alice", 0)], ["person", "separation"])
# 执行一轮BFS遍历
next_level_nodes = breadth_first_search(initial_nodes)
next_level_nodes.show()

• UDF仅适用于行级计算（比如给每行字符串做拼接、数值做运算），不能用来执行全局的DataFrame Join/Union/Spark SQL操作。
• 所有涉及SparkSession（spark.sql、spark.read等）的操作，都必须在Driver端执行，不能放到UDF、map/flatMap这类分布式执行的逻辑里。

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