data-pipeline-processor

用途

Data Pipeline Processor 是一个用于将数据文件通过转换管道进行处理的技能，核心流程为：输入（CSV/Excel/JSON）→ 验证 → 转换 → 分析 → 导出。它适用于需要以可配置、可重复的方式对数据文件进行清洗、转换和导出的场景。

主要能力

数据读取与编码检测

DataReader 类支持读取 CSV、xlsx、xls、json、parquet 格式。对于 CSV 文件，它会使用 chardet 检测编码，并按检测到的编码、utf-8、latin-1、cp1252 的顺序尝试解码，以处理编码问题。

数据验证

DataValidator 类支持添加可配置的验证规则，内置规则包括：

• required_columns_rule：检查必填列是否存在
• no_duplicates_rule：检查重复行
• non_null_rule：检查指定列是否含空值

验证结果包含 is_valid、errors、warnings 和 stats。

数据转换

DataTransformer 类提供链式转换操作：重命名列、按表达式过滤行、选择列、删除列、填充空值（支持指定值或 method）、类型转换（datetime、numeric 等）、添加计算列、分组聚合、排序。

数据导出

DataExporter 类支持导出为 CSV、Excel、JSON、Parquet，并会自动创建输出目录。

管道编排

DataPipeline 类根据 PipelineConfig 配置编排整个流程：读取输入、设置并执行验证、应用转换、导出结果。若验证失败则返回失败状态和错误信息；若有警告则记录警告。

使用方式

YAML 配置驱动

通过 YAML 文件定义 input、validation、transformations、output 各段。示例配置包括基础清洗管道（重命名、过滤、类型转换、填充空值、排序）和聚合管道（分组聚合、重命名、排序）。

命令行运行

python -m data_pipeline config/pipelines/clean_data.yaml

支持覆盖输入输出路径：

python -m data_pipeline config/pipelines/clean_data.yaml \
    --input data/custom_input.csv \
    --output data/custom_output.csv

支持 dry-run 仅验证模式：

python -m data_pipeline config/pipelines/clean_data.yaml --dry-run

编程调用

可通过导入 DataPipeline 和 PipelineConfig 以编程方式构建和运行管道。

适用场景

• 处理 CSV/Excel/JSON 文件并进行验证
• 数据清洗和转换工作流
• 批量文件处理与聚合
• 处理编码问题（UTF-8、Latin-1 回退）
• ETL（提取、转换、加载）操作
• 数据质量检查与报告

限制

资料中未提供关于性能上限、并发处理能力、支持的最大文件大小、依赖版本要求或平台兼容性的具体说明。