OpenMetadata 新连接器脚手架指南基于 scaffold-connector 的 Schema-First 开发全流程【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata导读本文系统讲解 OpenMetadata 仓库中内置的scaffold-connector命令与配套 Skill如何以一份 JSON Schema 为唯一事实源Schema-First的方式从零为 OpenMetadata 生成一个新连接器database / dashboard / pipeline / messaging / mlmodel / storage / search / api 等服务类型的完整骨架代码、测试连接定义与 AI 实现上下文。读完本文你将掌握metadata scaffold-connector的交互式与非交互式用法、连接器目录结构与注册清单以及从环境准备、分类、实现、注册、代码生成、静态校验、本地 Docker 联调到提交 PR 的九个完整阶段。一、入口与定位scaffold-connector 命令是什么scaffold-connector是 OpenMetadata 提供的一个 CLI 命令用于脚手架生成Scaffold一个全新的连接器。它在仓库中有三层载体命令定义skills/commands/scaffold-connector.md 是一份极简的命令说明书它本身不包含逻辑而是将执行委托给 Connector Building Skillopenmetadata-skills:scaffold-connector并约定若用户随命令提供了连接器名称或描述参数则将其透传给 Skill否则由 Skill 通过交互式提示引导完成。Skill 工作流skills/connector-building/SKILL.md 定义了从 Phase 0 到 Phase 9 的完整开发流程是本文的主体骨架。CLI 实现ingestion/src/metadata/cli/scaffold.py 实现了交互式收集、JSON Schema 生成、Python 模板生成与CONNECTOR_CONTEXT.md生成的全部逻辑命令入口注册在 ingestion/src/metadata/cmd.py。此外仓库还提供了一个薄封装脚本 scripts/scaffold_connector.py在metadataCLI 未安装时可直接运行python scripts/scaffold_connector.py其内部将ingestion/src加入sys.path后等价于执行metadata scaffold-connector。核心思想一份 JSON Schema 级联六层连接器开发遵循Schema-First 架构见 skills/connector-building/standards/main.md定义一份 JSON Schema随后的六层产物全部由此生成或受其约束JSON Schema唯一事实源 ├── Python Pydantic 模型 (make generate) ├── Java 模型 (mvn install -pl openmetadata-spec) ├── TypeScript 类型 (yarn parse-schema) ├── UI 配置表单 (RJSF 依据 Schema 自动渲染) ├── API 请求校验 (服务端使用 Java 模型) └── 测试夹具 (测试导入 Pydantic 模型)因此在脚手架阶段得到的连接 JSON Schema即单一事实源是整个连接器的地基后续不应手写配置类。二、Phase 0环境准备在执行任何make或python命令前先从仓库根目录创建并激活 Python 开发环境python3.11 -m venv env source env/bin/activate make install_dev generate后续每个阶段运行命令前都应确保环境已激活source env/bin/activate。make install_dev会安装开发依赖若缺失make py_format等格式化命令会失败make generate用于从已有 Schema 生成 Pydantic 模型。三、Phase 1生成脚手架Scaffold交互式模式source env/bin/activate metadata scaffold-connector交互模式会依次收集以下信息对应 scaffold.py 中的collect_interactive()连接器名称snake_case如my_db必须以小写字母开头仅允许小写字母、数字、下划线显示名称Display name默认按名称自动转为 CamelCase服务类型Service Typedatabase/dashboard/pipeline/messaging/mlmodel/storage/search/api连接类型仅 databasesqlalchemy默认最常见/rest_api如 Salesforce/sdk_client厂商 SDK非 database 类型仅提供rest_api与sdk_client认证类型basic/iam/azure/jwt/token/oauth可多选默认basic能力Capabilitiesmetadata始终、lineage、usage、profiler、stored_procedures、data_diff注意只有 database sqlalchemy 组合才会自动生成 lineage/usage/profiler 相关文件REST/SDK 类连接器默认仅metadata源码文档信息供 AI 上下文API/SDK 文档 URL、Python SDK 包名PyPI 名如boto3、关键 API 端点如GET /api/v1/databases, GET /api/v1/tables、额外的文档备注认证细节、分页、限流、特殊类型等支持多行输入集成测试信息可供 testcontainers 集成测试使用的 Docker 镜像如metabase/metabase:latest及要暴露的容器端口。非交互式模式metadata scaffold-connector \ --name my_db \ --service-type database \ --connection-type sqlalchemy \ --scheme mydbpymydb \ --auth-types basic \ --capabilities metadata lineage usage profiler \ --docs-url https://docs.example.com/api \ --sdk-package mydb-sdk \ --docker-image mydb/mydb:latest \ --docker-port 5432生成产物脚手架输出四类内容见 scaffold.py 模块 docstring连接 JSON Schema唯一事实源位于openmetadata-spec/.../entity/services/connections/{service_type}/{moduleName}Connection.json。生成器根据服务类型调用不同的属性构建函数database sqlalchemy_add_database_sqlalchemy_props含schemeSQLAlchemy 驱动枚举、username、authTypeoneOf引用./common/basicAuth.json等、token、hostPort、databaseName、databaseSchema、sslConfig、connectionOptions、connectionArguments、三个过滤模式schemaFilterPattern/tableFilterPattern/databaseFilterPattern以及依据能力开关生成的supportsMetadataExtraction、supportsProfiler、supportsDataDiff、supportsUsageExtraction、supportsLineageExtraction、supportsDBTExtraction、supportsQueryComment等标志均$ref自connectionBasicType.json的 definitionsdashboardhostPortformat: uriexpose: true、认证字段、dashboardFilterPattern/chartFilterPattern/projectFilterPatternpipelinehostPort、认证字段、pipelineFilterPatternmessagingbootstrapServers逗号分隔必填、认证字段、topicFilterPatternmlmodel / storage / search / api走通用属性模板hostPort 认证字段 supportsMetadataExtraction。测试连接 JSON按服务类型生成不同的校验步骤CheckAccess为公共必填第一步例如 database 追加GetSchemas、GetTables必填与GetViews可选若声明了 usage/lineage 能力还会追加GetQueriesdashboard 追加GetDashboards、GetChartspipeline 追加GetPipelines、GetPipelineStatusmessaging 追加GetTopicsmlmodel 追加GetModelsstorage 追加GetContainerssearch 追加GetSearchIndexesapi 追加GetCollections。Python 文件骨架/模板SQLAlchemy database 连接器生成可直接落地的代码模板connection.py、metadata.py、service_spec.py、queries.py、lineage.py、usage.py等其余类型生成带参考连接器指引的 skeleton 文件。CONNECTOR_CONTEXT.md作为任何 AI 工具Claude Code、Cursor、Codex、Copilot、Windsurf的实现工作文档位于连接器目录下。它已被gitignore仅保留在本地、绝不提交因此无需清理。四、Phase 2分类CLASSIFY脚手架按三个维度分类生成后应逐一核对维度一 — 服务类型决定目录位置与基类服务类型基类参考连接器databaseCommonDbSourceServicemysql/dashboardDashboardServiceSourcemetabase/pipelinePipelineServiceSourceairflow/messagingMessagingServiceSourcekafka/mlmodelMlModelServiceSourcemlflow/storageStorageServiceSources3/searchSearchServiceSourceelasticsearch/apiApiServiceSourcerest/BASE_CLASS_MAP与REFERENCE_CONNECTORS均定义于 scaffold.py非 SQLAlchemy 的 database 连接器改用DatabaseServiceSource参考 Salesforce。维度二 — 连接类型仅 databasesqlalchemy→BaseConnection[Config, Engine] SQLAlchemy dialectrest_api→get_connection() 自定义 REST 客户端参考salesforce/sdk_client→get_connection() 厂商 SDK 封装。维度三 — 能力metadata始终、lineage、usage、profiler、stored_procedures、data_diff决定额外生成哪些文件如lineage.py、query_parser.py。每个连接器位于ingestion/src/metadata/ingestion/source/{service_type}/{name}/目录内文件与用途如下Always 为必须文件用途是否必需__init__.py模块标记Alwaysconnection.py创建与测试连接Alwaysmetadata.py从数据源抽取元数据Alwaysservice_spec.py向框架注册连接器Alwaysclient.pyREST/SDK 客户端封装非 databasequeries.pySQL 查询模板databaselineage.py血缘抽取声明 lineage 能力时usage.py用量抽取声明 usage 能力时query_parser.py查询日志解析声明 lineage 或 usage 时CONNECTOR_CONTEXT.mdAI 实现简报由脚手架生成每种服务类型的详细模式见 skills/connector-building/standards/source_types/ 下对应文档。五、Phase 3调研RESEARCH读取脚手架生成的CONNECTOR_CONTEXT.md然后调研数据源的 API/SDK可派发子代理Claude Code启动connector-researcher代理Agent: openmetadata-skills:connector-researcher Prompt: Research {source_name} for an OpenMetadata {service_type} connector. Find: API docs, auth methods, key endpoints, pagination, rate limits, SDK packages.不可派发子代理使用 WebSearch / WebFetch 自行调研。调研关注点API 文档、认证方式、关键端点、分页机制、限流策略、SDK 包。Skill 的 examples 目录提供了三份可直接参考的调研样板含端点与备注分别对应三条典型路径database-sqlalchemy.yamlClickHouse 型 SQLAlchemy 连接器schemeclickhousedbconnect、默认端口 8123、能力含 lineage/usage/profiler/data_diff、备注系统库排除清单与system.query_log查询日志表dashboard-rest.yamlApache Superset 型 REST 连接器GET /api/v1/dashboard/等端点、basic 登录换取 JWT、page/page_size 分页、dashboard→chart→dataset 的层级关系pipeline-sdk.yamlPrefect 型 SDK 连接器prefect-client包、Bearer token、Flows管线 / Flow Runs执行、offset/limit 分页、状态映射 COMPLETEDSuccessful 等。六、Phase 4实现IMPLEMENT脚手架生成的文件带有# TODO标记实现前应先阅读相关标准文档见 skills/connector-building/standards/connection.md连接模式、patterns.md错误处理、分页、认证、performance.md分页、查询优化、反模式、memory.md内存管理与流式、source_types/{service_type}.md服务专属模式。SQLAlchemy database模板基本完整通常只需按需定制_get_client()。脚手架生成的connection.py默认实现见 scaffold.py 的gen_connection_database_sqlalchemy已经给出标准写法——继承BaseConnection[Config, Engine]_get_client()调用create_generic_db_connectionget_connection_url_commonget_connection_args_common构建引擎test_connection()委托给test_connection_db_schema_sourcesmetadata.py的create()则用WorkflowSource.model_validate解析配置并校验连接类型为{Camel}Connection否则抛出InvalidSourceExceptionservice_spec.py依据能力声明自动装配DefaultDatabaseSpec含metadata_source_class、可选lineage_source_class/usage_source_class、connection_class。非 SQLAlchemy研究参考连接器后逐个实现 skeleton 文件。关键实现红线JSON Schema需要默认认证的服务必须将认证字段username、password、token设为required——若省略某字段在运行时只会得到不透明的 401设为必填可在 UI 表单阶段就完成校验所有走 HTTPS 通信的连接器都必须包含 SSL/TLS 配置verifySSLsslConfig$ref企业部署常用内部 CASSL 必须端到端打通schema →connection.py用get_verify_ssl_fn解析→client.pysession.verify verify_ssl。接线缺失会导致 SonarQube 安全审查失败。Pydantic API 模型models.py凡使用Field(alias...)必须设置model_config ConfigDict(populate_by_nameTrue)否则用 Python 属性名构造实例会抛ValidationError。非 database 连接器client.pyAPI 支持分页的列表端点必须实现分页否则会出现静默丢数据只摄取第一页对高频查找如文件夹路径→文件夹名应构建 dict 而非反复遍历列表。存储类与任何读取文件的连接器禁止不经大小检查就.read()整个文件生产环境会 OOM数据文件应使用框架流式读取器metadata/readers/dataframe/大对象处理后del并调用gc.collect()。血缘lineage严禁在搜索查询中使用通配符table_name*——这会把库中每张表都关联到每个实体产生错误血缘若数据源不提供表级信息宁可跳过血缘并在文档中说明该限制。七、Phase 5注册REGISTER连接器代码生成后还需修改八个集成点详见 skills/connector-building/standards/registration.md遗漏任何一个都会导致连接器不出现在 UI、使用默认图标或驱动安装失败步骤文件改动1openmetadata-spec/src/main/resources/json/schema/entity/services/{serviceType}Service.json向serviceType枚举添加类型并在config的oneOf中$ref连接 Schema2ingestion/setup.py添加连接器 pip extras 块保持字母序使pip install openmetadata-ingestion[mydb]与 Docker 摄取镜像能拉到 SDK/驱动3ingestion/src/metadata/examples/workflows/{name}.yaml可运行的 CLI 工作流示例metadata ingest -c供用户复制部署与 QA 冒烟测试4openmetadata-ui/.../utils/{ServiceType}ServiceUtils.tsx通过loadConnectionSchema动态加载连接 SchemaJSON 由yarn parse-schema生成到public/jsons/connectionSchemas/随改动一并提交并注册 loader5openmetadata-ui/.../assets/img/service-icon-{name}.png添加服务 logo 资产优先 SVG否则 ≥128×128 方形 PNG6openmetadata-ui/.../utils/ServiceIconUtils.ts导入资产并注册进SERVICE_ICON_LOADERS查找键为服务类型小写并去除_/-getServiceIcon()用toLowerCase().replaceAll(/[_-]/g, )归一化7openmetadata-ui/.../public/locales/en-US/{ServiceType}/{Name}.md表单字段级帮助文档用$$section ... $$块按 Schema 字段id组织8openmetadata-ui/.../constants/ServiceType.constant.ts追加到BETA_SERVICES—— 新连接器一律以 Beta 状态发布待生产验证后才在后续 PR 中移除其中第 3 步的 YAML 模板结构source/sink/workflowConfig三段应与其他连接器保持一致仅source.*块因连接器而异jwtToken处可使用开发用 JWT 占位保证开箱即用。明确无需改动i18n locale 文件显示名来自生成的{ServiceType}Type枚举Services.constant.ts已废弃仅是 re-export 垫片图标注册请到ServiceIconUtils.tssrc/generated/下的生成类型由yarn parse-schema产出禁止手改。八、Phase 6代码生成与格式化GENERATE FORMAT提交前必须执行不可跳过未格式化的代码会被 CI 拒绝# 确保环境激活、工具已安装 source env/bin/activate pip install -e .[dev] 2/dev/null || make install_dev # 依据 Schema 生成模型 make generate # Python Pydantic 模型 mvn clean install -pl openmetadata-spec # Java 模型 cd openmetadata-ui/src/main/resources/ui yarn parse-schema # UI Schemas # 格式化全部代码提交前强制 cd /path/to/repo/root make py_format # ruff lint-fix format mvn spotless:apply # 格式化 Java若make py_format失败最常见原因是缺少开发依赖——先运行make install_dev再重试。九、Phase 7静态校验与自查清单VALIDATE提交前用仓库自带的静态分析器自检python skills/connector-review/scripts/analyze_connector.py {service_type} {name}修复其报告的问题后逐项核对完整清单原文见 SKILL.md 第 203–223 行[ ] JSON Schema: 可校验、$ref 可解析、supports* 标志正确 [ ] JSON Schema: 服务强制认证时认证字段为 required [ ] JSON Schema: HTTPS 连接器包含 SSL/TLS 配置 [ ] 代码生成: make generate mvn install yarn parse-schema 全部成功 [ ] 连接: 可创建 clienttest_connection 所有步骤通过 [ ] Source: create() 校验配置类型ServiceSpec 可被发现 [ ] Pydantic 模型: 所有带 alias 的模型设置 populate_by_nameTrue [ ] Client: 所有列表端点实现分页以 API 文档为准 [ ] Client: prepare() 中使用 dict 查找而非逐实体遍历列表 [ ] 血缘: 无通配符 table_name*无表级信息时跳过 [ ] 测试: 单元 连接集成 元数据集成测试通过无空桩 [ ] 注册: 服务 schema、setup.py、{ServiceType}ServiceUtils.tsx 均已更新 [ ] 注册: ingestion/src/metadata/examples/workflows/ 中有 CLI 工作流示例 [ ] 注册: 服务图标资产存在且注册于 ServiceIconUtils.ts [ ] 注册: public/locales/en-US/{ServiceType}/{Name}.md 文档就位 [ ] 注册: 服务类型已追加到 ServiceType.constant.ts 的 BETA_SERVICES [ ] 格式化: make py_format mvn spotless:apply 通过且无变更 [ ] 清理: CONNECTOR_CONTEXT.md 未被 gitignore确认未被暂存 [ ] 清理: 无遗留 TODO 脚手架注释十、Phase 8本地 Docker 联调TEST LOCALLY构建全部组件并用 Docker 拉起完整本地 OpenMetadata 栈脚本见 docker/run_local_docker.sh全量构建首次或改动 Java/UI 后./docker/run_local_docker.sh -m ui -d mysql -s false -i true -r true快速重建仅改 ingestion约 2–3 分钟./docker/run_local_docker.sh -m ui -d mysql -s true -i true -r false服务就绪后约 3–5 分钟打开http://localhost:8585进入Settings → Services → {你的服务类型}点击Add New Service并选择你的连接器填写连接信息并点击Test Connection测试通过后运行 metadata ingestion验证实体是否被创建。其他服务地址Airflow http://localhost:8080admin / admin、Elasticsearch http://localhost:9200。拆除cd docker/development docker compose down -v排障连接器不在下拉列表 → 检查服务 schema 注册重建时去掉-s true测试连接失败 → 检查test_fn的 key 是否与测试连接 JSON 的步骤名一致查看容器日志docker compose -f docker/development/docker-compose.yml logs ingestion十一、Phase 9提交 PRCREATE PR提交 PR 时建议将静态分析器的质量评估摘要写入 PR 描述让维护者无需逐文件评审即可了解连接器状态# 运行静态分析器 analysis$(python skills/connector-review/scripts/analyze_connector.py {service_type} {name} --json) # 创建 PR描述中包含质量摘要 gh pr create --title feat(ingestion): Add {Name} {service_type} connector --body $(cat EOF ## Summary - New {service_type} connector for {Name} - Capabilities: {list capabilities} ## Test plan - [ ] Unit tests pass (pytest ingestion/tests/unit/topology/{service_type}/test_{name}.py) - [ ] Integration tests pass - [ ] Local Docker test: connector appears in UI, test connection passes ## Connector Quality Review **Verdict**: {VERDICT} | **Score**: {SCORE}/10 | Category | Score | |----------|-------| | Schema Registration | X/10 | | Connection Auth | X/10 | | Source, Topology Performance | X/10 | | Test Quality | X/10 | | Code Quality Style | X/10 | **Blockers**: 0 | **Warnings**: {count} | **Suggestions**: {count} details summaryStatic analysis output/summary {paste analyze_connector.py output here} /details EOF )质量摘要让维护者能快速建立信心也便于后续提交connector-review 类 Skill 与 skills/connector-review/scripts/analyze_connector.py 可复用同一套评估口径。十二、标准文档速查实现过程中涉及的全部标准位于 skills/connector-building/standards/标准内容main.md架构总览、连接器解剖、服务类型patterns.md错误处理、日志、分页、认证、过滤器testing.md单元测试模式、集成测试、pytest 风格code_style.mdPython 风格、JSON Schema 约定、命名schema.md连接 Schema 模式、$ref 用法、测试连接 JSONconnection.mdBaseConnection 与函数模式、SSL、客户端封装service_spec.mdDefaultDatabaseSpec 与 BaseSpecregistration.md服务枚举、UI utils、i18nperformance.md分页、批量处理、限流memory.md内存管理、流式、OOM 防护lineage.md血缘抽取方法、方言映射、查询日志sql.mdSQLAlchemy 模式、URL 构建、认证、多数据库source_types/*.md各服务类型专属模式架构级参考资料位于 skills/connector-building/references/architecture-decision-tree.md服务类型/连接类型/基类选择、connection-type-guide.mdSQLAlchemy vs REST vs SDK、capability-mapping.md按服务类型的能力映射与 Schema 标志。结语scaffold-connector把 OpenMetadata 连接器开发从阅读大量参考实现后手工复制压缩为一次脚手架生成 九阶段引导。其关键价值在于 Schema-First连接 JSON Schema 一经生成Python/Java/TypeScript 模型、UI 表单、API 校验与测试夹具全部随之而来而CONNECTOR_CONTEXT.md又让 AI 工具可以带着完整上下文直接进入实现阶段。只要严格走完分类核对、注册八步、代码生成与格式化、静态校验与本地 Docker 联调一个新连接器就能以 Beta 状态安全地进入 OpenMetadata 生态。【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考