本文介绍如何将配置信息存放在独立的 python 文件中，并在另一文件中导入该配置，动态提取字段名用于 dataframe 数据校验。
本文介绍如何将配置信息存放在独立的 python 文件中，并在另一文件中导入该配置，动态提取字段名用于 dataframe 数据校验。
在实际数据处理项目中，将配置（如字段列表、校验规则、元数据）与业务逻辑分离是推荐实践。这不仅提升可维护性，还便于多环境适配和配置复用。本文以 config.py 存储结构化配置、validator.py 实现动态字段校验为例，完整演示跨文件配置读取与应用。
✅ 步骤一：保存配置到独立 Python 文件
将您的字典保存为 config.py（注意：不加 .py 后缀在 import 时无需写）：
# config.py
dict_nova = {
"export_nova_rosterlight": {
"metadata_resource_id": "export_nova_rosterlight",
"metadata_product_version": "1.0",
"List_Fields": [
"data_export",
"userID",
"deviceName",
"osVersion",
"emailAddress",
"model",
"serialNumber"
]
},
"export_nova_export": {
"metadata_resource_id": "Export_nova_export",
"metadata_product_version": "1.0",
"List_Fields": [
"data_export",
"userMail",
"userID",
"deviceName",
"emailAddress"
]
}
}⚠️ 注意：config.py 应置于与使用它的脚本同一目录，或确保其路径已加入 sys.path；避免命名为 config.py 以外的常见模块名（如 json.py, datetime.py），以防命名冲突。
python-script-generator下载快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
✅ 步骤二：在主逻辑文件中导入并使用配置
在 validator.py（或其他业务文件）中，导入 config 模块并重构 validate_counts 函数，使其支持按配置键动态选取字段：
立即学习“Python免费学习笔记（深入）”；
# validator.py
from pyspark.sql import functions as F # 假设使用 PySpark；如为 Pandas，请替换为相应语法
import config # ← 关键：直接导入 config 模块
def validate_counts(df, config_key="export_nova_rosterlight"):
"""
根据 config.py 中指定 key 的 List_Fields 动态校验列存在性与统计逻辑
:param df: 输入 DataFrame
:param config_key: config.dict_nova 中的键名，如 "export_nova_rosterlight"
:return: 'valid' 或 'invalid'
"""
# ✅ 安全获取配置，防止 KeyError
if config_key not in config.dict_nova:
raise ValueError(f"Config key '{config_key}' not found in config.dict_nova")
fields = config.dict_nova[config_key]["List_Fields"]
# ✅ 验证必需字段是否存在（增强鲁棒性）
missing_cols = [col for col in ["emailAddress", "deviceName"] if col not in df.columns]
if missing_cols:
raise ValueError(f"Missing required columns: {missing_cols}")
# 执行原始校验逻辑
distinct_mail = df.agg(F.countDistinct('emailAddress')).collect()[0][0]
distinct_device = df.agg(F.countDistinct('deviceName')).collect()[0][0]
if 10000 <= distinct_mail <= 20000 and 5000 <= distinct_device <= 10000:
return 'valid'
else:
return 'invalid'
# ✅ 使用示例
# from pyspark.sql import SparkSession
# spark = SparkSession.builder.getOrCreate()
# df = spark.read.csv("data.csv", header=True)
# result = validate_counts(df, "export_nova_rosterlight")
# print(result) # 输出 'valid' 或 'invalid'✅ 进阶建议
- 类型安全 & IDE 支持：可将 config.py 改为使用 dataclass 或 TypedDict（Python 3.8+），配合类型提示提升可读性与静态检查能力；
- 配置热加载：如需运行时更新配置，可用 importlib.reload(config)（慎用，注意副作用）；
-
替代方案对比：
- ✅ .py 配置：支持复杂逻辑、注释、条件分支，适合中大型项目；
- ⚠️ JSON/YAML：更轻量、语言无关，但无法执行代码逻辑；
- ❌ 环境变量：适合简单键值，不适用于嵌套结构体。
通过以上方式，您实现了配置与逻辑解耦、字段来源可配置化、校验逻辑可复用——这是构建健壮数据管道的重要一步。
