如何通过读取MongoDB集合配置动态调用对应Python处理函数

第一步：依赖安装与MongoDB连接准备 #

首先安装MongoDB Python驱动：

pip install pymongo

初始化MongoDB连接，替换为你实际的数据库地址、库名、集合名：

import pymongo
from pyspark.sql.functions import regexp_replace, lower, expr

# 初始化MongoDB连接
client = pymongo.MongoClient("mongodb://你的MongoDB地址:端口/")
db = client["你的业务库名"]
config_col = db["attributes"]

第二步：读取MongoDB全量配置 #

直接拉取集合内所有配置，建议转成属性名为key的字典方便后续查询：

# 读取全量配置
config_dict = {}
for config_item in config_col.find():
    config_dict[config_item["attributes_names"]] = {
        "data_type": config_item["attributes_datype"],
        "allow_null": config_item["attributes_isNull"] == "true",
        "process_func_name": config_item["attributes_std_function"]
    }

第三步：注册可调用函数映射 #

不要直接用eval执行配置里的函数名，存在注入风险，提前把所有允许调用的处理函数注册到映射表中：

# 业务处理函数实现
def email_valid(df):
    df1 = df.withColumn(df.columns[0], regexp_replace(lower(df.columns[0]), "^a-zA-Z0-9@\._\-| ", ""))
    extract_expr = expr(
        "regexp_extract_all(emails, '(\\w+([\\.-]?\\w+)*@\\[A-Za-z\-\.]+([\\.-]?\\w+)*(\\.\\w{2,3})+)', 0)")
    df2 = df1.withColumn(df.columns[0], extract_expr) \
        .select(df.columns[0])
    return df2

def address_valid(df):
    # 地址校验逻辑自行实现，结构和email_valid一致即可
    pass

# 函数映射表，新增处理函数只需要在这里加映射即可
func_map = {
    "email_valid": email_valid,
    "address_valid": address_valid
}

第四步：按配置调用对应函数 #

遍历待处理的字段数据，匹配配置调用对应函数即可：

# 待处理数据示例：key为属性名，value为对应待处理的DataFrame
field_df_map = {
    "email": email_raw_df,
    "address": address_raw_df
}

process_result = {}
for field_name, raw_df in field_df_map.items():
    # 匹配配置
    field_config = config_dict.get(field_name)
    if not field_config:
        print(f"字段{field_name}无匹配配置，跳过处理")
        continue
    # 校验处理函数是否已注册
    target_func = func_map.get(field_config["process_func_name"])
    if not target_func:
        print(f"字段{field_name}配置的函数{field_config['process_func_name']}未注册，跳过处理")
        continue
    # 非空校验逻辑（可选）
    if not field_config["allow_null"]:
        null_count = raw_df.filter(raw_df[raw_df.columns[0]].isNull()).count()
        if null_count > 0:
            raise ValueError(f"字段{field_name}存在空值，不符合非空配置要求")
    # 调用对应函数处理
    process_result[field_name] = target_func(raw_df)

注意事项 #

• 后续新增处理规则仅需要三步：实现处理函数、在func_map中注册映射、在MongoDB中新增对应配置即可，不需要修改核心调度逻辑
• 可以提前加配置合法性校验，读取配置后先检查所有配置的函数是否都在映射表中，提前发现配置错误
• 如果需要处理非DataFrame类型的参数，可以自行调整函数传参逻辑，适配不同业务场景

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