OpenDataPipeline 是一款 AI 驱动的数据迁移与数据转换工具（data migration & data transformation），内置统一的数据源连接层，一次配置即可连接 MySQL、PostgreSQL、Snowflake、Databricks 等 12+ 种数据库与数据源，帮助新手快速完成跨数据库的数据集成。

本文将从「支持哪些数据库 → 连接机制 → 四大数据库配置实战 → 本地部署 → 常见问题」带你从零打通第一条数据管道。

🧭 支持的数据源清单：12+ 种数据库一览

OpenDataPipeline 的连接器注册中心通过工厂模式统一管理数据源类型（见 core/datasource/connector.py），目前内置支持的数据源如下：

💡 除 SQL 数据库外，它还支持 S3 对象存储和 Google Sheets，非常适合「云存储 + 数仓 + 关系库」混合场景的数据迁移。

完整的连接组件说明可参考 core/datasource/README.md。

🔍 连接机制速览：YAML 驱动 + 工厂注册

OpenDataPipeline 的数据源连接层由三部分组成，理解它能让后续排查问题事半功倍：

1. 数据库 Profile（YAML 配置）：每种数据库一个 YAML 文件，声明驱动名、默认端口、版本查询语句、JDBC 模板等，目录位于core/datasource/db_profiles/；
2. 注册中心：core/datasource/registry.py 自动加载所有 Profile 并合并默认值，新增标准 RDBMS只需加一个 YAML 文件，无需改 Python 代码；
3. 工厂类：DataSourceConnector根据类型字符串（如postgres、databricks）懒加载对应实现类，避免一次性加载所有依赖。

而界面上「连接时该填哪些输入框」则由 opendatapipeline/src/configurations/datasources/datasources.yml 定义——每个数据源的图标、分类、参数表单（Host、Port、Username、Password 等）都在这里声明。

连接成功后，数据源信息会通过 opendatapipeline/src/models/mongo/mongo_factory.py 持久化到 MongoDB，供后续管道任务复用。

🐬 MySQL 连接配置实战

MySQL 是典型的「Host + Port + 库名」型数据源，配置最简单。

表单参数（定义于 datasources.yml 中driver: mysql节点）：

• Username / Password：数据库账号
• Hostname：主机地址，如localhost
• Port：默认3306（来自 mysql.yml 的default_port）
• Database：目标库名

连接层使用mysql+mysqlconnector驱动，Spark 侧自动注入mysql-connector-java的 JDBC 驱动，无需手动打包。

⚡ 进阶技巧：在连接参数中传入{"aod_schema": ["schema1"]}、{"aod_table": ["t1","t2"]}可以只加载指定 schema/表，大幅提升大数据量下目录列表的速度。

🐘 PostgreSQL 连接配置实战

PostgreSQL 与 MySQL 参数结构一致，注意默认端口为5432（见 postgres.yml）。

• 驱动：postgresql+psycopg2（Python 侧）+org.postgresql.Driver（Spark JDBC 侧）
• 额外能力：Profile 中内置了 dlt 目标端优化参数，如executemany_batch_page_size: 10000，作为 DMS（数据迁移）目标端时批量写入性能更好。

支持pool_size、max_overflow、pool_timeout、pool_recycle等 SQLAlchemy 连接池选项，高频读写场景建议调大pool_size。

❄️ Snowflake 连接配置实战

Snowflake 的参数与普通关系库不同，没有 Port，取而代之的是 Account Identifier 和 Warehouse：

1. Username / Password：账号密码
2. Account Identifier：账户标识，格式为orgname-account_name，例如helical-NM101（填入host字段）
3. Database：数据库名
4. Warehouse：执行查询的虚拟计算集群，必须处于 running 状态

连接建立后会自动执行USE WAREHOUSE {warehouse}校验；如果忘记填 Warehouse，系统会给出明确提示：Please edit the DataSource and add Warehouse details.（见 snowflake.yml 的pre_query_commands）。

⚠️ 若使用 dlt 迁移（DMS 模式），Snowflake 走原生目的地（dlt_destination: native），比 SQL 通道更快。

💠 Databricks 连接配置实战

Databricks 是「URL 直连 + Token」模式，不需要端口：

• Host：Workspace 完整域名，例如dbc-xxxx.cloud.databricks.com
• HTTP Path：数据库的 JDBC HTTP Path
• Access Token：在 Databricks 控制台生成的访问令牌（填入 password 字段）

底层自动拼装为jdbc:databricks://{host}:443/并使用databricks-jdbc驱动（见 databricks.yml）。由于 Databricks 目录结构特殊，配置中开启了requires_catalog: true，元数据通过SHOW TABLES IN {catalog}.{schema}获取，连接时请确保 Token 对目标 catalog 有读权限。

🗂️ 本地快速部署：先跑起来再连库

推荐用 Docker 一键拉起整套环境，数据库连接配置在 Web 界面中即可完成，无需手动写代码：

部署入口文件：docker/docker-compose.yml，本地初始化脚本：odp_open_source_local_setup.sh。启动后在 Web 界面选择数据源类型、填写对应表单参数、点击测试连接即可。

✅ 常见连接问题排查清单

📚 相关文件速查

• 连接组件总览：core/datasource/README.md
• 工厂与注册：core/datasource/connector.py、core/datasource/registry.py
• 数据库 Profile 配置：core/datasource/db_profiles/
• UI 数据源表单定义：opendatapipeline/src/configurations/datasources/datasources.yml
• 部署编排：docker/docker-compose.yml

掌握以上配置后，你就可以在 OpenDataPipeline 中自由组合任意「源库 → 目标库」的迁移与转换管道，把 12+ 种数据源真正用起来。