从CSV到数据库只需几步：OpenDataPipeline平文件数据导入操作指南（含Excel、自定义分隔符处理）-平芜编程栈

【免费下载链接】open-data-pipeline

Open Data Pipeline is an AI powered data migration and data transformation tool

项目地址：https://gitcode.com/gh_mirrors/op/open-data-pipeline

OpenDataPipeline（Open Data Pipeline）是一款 AI 驱动的数据迁移与数据转换工具，支持将 CSV、Excel 等平文件（Flat Files）快速导入数据库，并可自定义分隔符、类型推断、数据预览。本文将带你用几步完成平文件数据导入，覆盖 Excel 读取与分号分隔符等特殊场景。

一、先认识平文件导入能力

OpenDataPipeline 的平文件导入由三层协作完成：

上传层：文件上传后保存到你的工作区sources/flat_files目录，并记录元数据（文件名、类型、路径），见 files/service.py
读取层：底层基于 PySpark 读取文件，CSV 默认开启表头识别和 Schema 推断（header: True、inferSchema: True），列名会自动清理，见 read.py
管道层：通过read_files函数把文件读入 DataFrame，再经转换后导出到目标表，入口在 extractor.py

二、准备工作：一键安装步骤

本地体验只需满足版本要求并运行一键脚本：

安装Docker v24+与Node v18+

克隆仓库：

git clone https://gitcode.com/gh_mirrors/op/open-data-pipeline

在仓库根目录执行一键部署脚本 odp_open_source_local_setup.sh：
```
bash odp_open_source_local_setup.sh
```
脚本会自动校验 Docker/Node 版本、构建前端并拉起容器服务。
若使用 Windows + Docker Desktop，注意 WSL 2 引擎配置（如上图），保证容器网络可用。

💡 上传的文件默认大小限制为5MB，可在设置页调整（files/service.py#L120-L122）。

三、导入 CSV：从上传到落库 3 步

第 1 步：上传平文件

在聊天工作台（Chat）中上传你的.csv文件，系统会生成file_id并在侧边栏显示文件卡片，可重命名、删除、点击预览。上传接口为 upload_file，前端卡片组件见 ADFile/index.jsx。

仓库中自带示例数据可直接上手，例如 Students.csv、Employees.csv。

第 2 步：读取文件为 DataFrame

上传后 AI 会自动编排一条read_files管道，等价配置如下：

- function: read_files output: dataframe_alias: students source_id: 669e3404fd5c32fb759d2b2c parameters: file_id: <你的file_id> file_name: Students

参数说明与完整示例见 yaml_documentation.md#L416-L438。

第 3 步：导出到数据库表

确认数据无误后，用export_table把 DataFrame 写入目标数据库表：

- function: export_table parameters: connection_id: <目标库连接ID> table_name: 'public.students' source_id: <上一步的source_id>

配置说明见 yaml_documentation.md#L543-L567。至此，CSV 数据已安全落库 ✅

四、导入 Excel（xlsx/xls）：和 CSV 一样简单

Excel 文件通过 Spark 的com.crealytics.spark.excel数据源读取，同样支持表头识别与 Schema 推断，实现见 read.py#L50-L75。

两点提示：

多 Sheet 处理：可用dataAddress参数精确指定工作表区域，例如"'Sheet1'!A1:B2"，用法参考 test_read.py#L56-L60
列筛选：读取时可通过columns参数只取需要的列，减少内存占用

测试样例 msheets.xlsx 可作为多 Sheet 参考数据。

五、自定义分隔符：分号分隔 CSV 怎么处理

很多从 ERP 系统导出的 CSV 用分号;而非逗号分隔。OpenDataPipeline 通过config参数传入delimiter即可解析，实现位于 read.py#L33-L39：

config = {'delimiter': ';'} dataframe = read.csv(spark, path, config=config)

对应测试与样例文件：

分号分隔样例：file1_with_semicolon.csv
分隔符 + 列筛选组合用法：test_read.py#L39-L44

⚠️ 常见问题：分号文件按默认逗号读取后只剩一列——这不是数据损坏，只需在读取配置中加上delimiter: ;即可。Tab 分隔同理，传delimiter: '\t'。

六、导入后常用转换：让数据更规范

落库前可对数据做轻量清洗，全部通过对话即可完成，常用函数清单（文档：yaml_documentation.md）：

需求	函数	效果
去重	`deduplicate`	按指定列删除重复行
改列名	`rename_columns`	适配数据库命名规范
类型转换	`typecast`	字符串转 date/int 等
补空值	`fill_na`/`drop_na`	填充或删除缺失值
多表合并	`joins`/`union`	按 key 关联或纵向拼接

七、常见问题 FAQ

上传提示 size 超限？默认 5MB，请在 Settings 中调大file_size限制。
列名含空格导致报错？读取层会自动调用clean_column_name清理列名（read.py#L40-L41），一般无需手动处理。
支持的文件类型？.csv、.xlsx、.xls为核心平文件类型（download/service.py#L62-L64），更大规模迁移可结合 S3 数据源使用。