PySpark中类内函数调用UDF及'devAM_hive'模块找不到问题求助

我来帮你拆解并解决这个问题，你遇到的ModuleNotFoundError: No module named 'devAM_hive'本质是Spark分布式执行特性+类方法写法不规范共同导致的，咱们一步步来处理：

问题根源 #

1. 类方法定义不严谨：你的aux方法是类内函数，但未声明为@staticmethod或@classmethod，直接通过Anomalie.aux调用会引发引用问题。
2. Spark分布式依赖缺失：Spark会把UDF序列化后分发到集群所有worker节点执行，但worker节点的Python环境默认没有你的devAM_hive模块，导致找不到依赖。

解决方案 #

1. 修正类内方法定义 #

先把aux声明为静态方法，确保它能被类直接调用，同时适配UDF的序列化要求：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StringType
import re

class Anomalie():
    def __init__(self):
        # 建议不在__init__中提前初始化UDF，避免序列化问题
        pass

    @staticmethod
    def aux(texte):
        code_utilisateur = re.findall(r'[\s]*\d{2}.\d{2}.\d{4}[\s]*\d{2}.\d{2}.\d{2}\s(\w?.?\s?.*)\s\(', texte)
        return code_utilisateur

    def auto_test(self, df):
        # 在使用时动态创建UDF，确保静态方法能被正确识别
        anomalie_udf = F.udf(Anomalie.aux, ArrayType(StringType()))
        df = df.withColumn("name", anomalie_udf(F.col("Description")))
        return df

2. 解决模块分发问题 #

要让worker节点能找到devAM_hive模块，推荐以下几种实用方法：

方法一：提交脚本时用--py-files参数

如果用spark-submit提交主脚本，直接指定devAM_hive.py，Spark会自动把模块分发到所有worker节点：

spark-submit --py-files devAM_hive.py your_main_script.py

方法二：主脚本中手动添加模块路径（适合本地/同目录场景）

如果devAM_hive.py和主脚本在同一目录，可在主脚本开头添加：

import sys
import os
# 获取当前脚本所在目录
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
    sys.path.append(current_dir)

from devAM_hive import *
# 后续调用代码不变
A = Anomalie()
df = A.auto_test(row_data)
df.select("name").show(50)

注：这种方式在集群环境下可靠性稍弱，优先推荐--py-files方案。

方法三：打包成egg/wheel文件（适合复杂项目）

如果你的代码是完整包结构，可打包成egg文件后提交：

# 打包成egg
python setup.py bdist_egg
# 提交任务
spark-submit --py-files dist/your_package-0.1.0-py3.8.egg your_main_script.py

额外小建议 #

如果你的UDF逻辑不依赖类的状态，也可以把aux方法提到类外部作为独立函数，这样UDF的定义和序列化会更简洁，还能避免类序列化的潜在问题。

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