企业官网建设流程全解析

OpenMetadata Custom Database 连接器完全指南：用自定义 Python 类接入任意数据源

【免费下载链接】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 的 Custom Database 连接器展开，讲解如何通过sourcePythonClass与connectionOptions将业务/工程环境中独有的数据源接入 OpenMetadata 元数据平台。读完本文，你将掌握自定义连接器的配置方式、底层类加载机制、参数传递方法与连接校验策略，并能参照仓库源码（customDatabaseConnection.json、ingestion.py）独立实现一个可运行的 Custom Database 数据源。

什么是 Custom Database 连接器

OpenMetadata 内置了大量数据库连接器（MySQL、PostgreSQL、Glue、SAS 等），但实际业务中往往存在 OpenMetadata 尚未内置、甚至外部厂商也不提供标准驱动的数据源。Custom Database 连接器正是为此设计的通用包装器：

它是任何你编写并添加到 OpenMetadata ingestion 镜像中的 Python 类的外层封装，核心思路是给你一套工具，把仅存在于自身业务/工程上下文中的任意数据源引入 OpenMetadata。

也就是说，连接器本身不做任何数据获取工作，真正干活的是你通过sourcePythonClass指定的 Python 类。OpenMetadata 的 ingestion workflow 负责按既定节奏实例化该类、调用其next_record方法不断拉取记录，并把记录以标准实体（如 Table）的形式写入 OpenMetadata API。

从连接类型定义看，Custom Database 属于 Database 服务下的独立连接类型，其 JSON Schema 定义在 customDatabaseConnection.json 中，javaType为org.openmetadata.schema.services.connections.database.CustomDatabaseConnection，Schema 描述为："Custom Database Service connection to build a source that is not supported by OpenMetadata yet."（用于构建 OpenMetadata 尚不支持的数据源）。

Connection Details：连接配置项详解

Custom Database 连接在 OpenMetadata UI（或通过 API/配置文件）中需要配置以下字段，均对应customDatabaseConnection.json中的 JSON Schema 属性。

Source Python Class（sourcePythonClass）

字段说明：需要被 ingestion workflow 实例化的 Source Python 类的完整类名。注意，这个类必须实现next_record方法，Workflow 才能持续读取记录并发送到 OpenMetadata API。

在 JSON Schema 中该字段定义为：

"sourcePythonClass": { "title": "Source Python Class Name", "description": "Source Python Class Name to instantiated by the ingestion workflow", "type": "string" }

配置示例：传入类的完整限定名（模块路径 + 类名），例如：

sourcePythonClass: "my_company.connectors.my_csv_connector.MyCSVSource"

仓库中已有实际用法，例如 SAS 连接器在构建 Custom Database 源时使用的类名（见 sas/metadata.py）：

sourcePythonClass="metadata.ingestion.source.database.customdatabase.metadata.SASDB",

Connection Options（connectionOptions）

字段说明：当需要向 Source 类传入输入参数时使用。例如希望根据名为business_unit的参数值运行某段逻辑，可以以 key-value 形式传入business_unit键，然后在 Source 类中读取它。

原文给出的读取方式（基于 pydantic v1 的__root__访问）：

business_unit = self.service_connection.connectionOptions.__root__.get("business_unit")

在 JSON Schema 中该字段引用connectionBasicType.json中定义的通用connectionOptions结构：

"connectionOptions": { "title": "Connection Options", "$ref": "../connectionBasicType.json#/definitions/connectionOptions" }

connectionOptions本质是一组自由键值对（key-value），OpenMetadata 不对其内容做类型约束，可用来传递任意业务参数：文件路径、过滤规则、认证信息片段、业务单元标识等。由于 Schema 声明了"additionalProperties": true，即使未来需要扩展自定义字段，也不会被模型校验拒绝。

其他 Schema 字段

除上述两个核心字段外，customDatabaseConnection.json 还定义了以下属性，配置时可一并使用：

其中三个过滤模式与 OpenMetadata 全局的 filterPattern 定义一致（引用type/filterPattern.json），可用于在自定义源上同样实现 schema/table/database 级别的包含排除控制。

底层原理：sourcePythonClass 是如何被加载与实例化的

Custom Database 之所以能加载"任意 Python 类"，关键在于 ingestion workflow 的动态导入机制。

Workflow 侧：按前缀走自定义导入分支

在 workflow/ingestion.py 的import_source_class方法中，逻辑明确区分了自定义连接器与内置连接器：

def import_source_class(self) -> type[Source]: source_type = self.config.source.type.lower() try: return ( import_from_module(self.config.source.serviceConnection.root.config.sourcePythonClass) if source_type.startswith(CUSTOM_CONNECTOR_PREFIX) else import_source_class(service_type=self.service_type, source_type=source_type) ) except DynamicImportException as e: if source_type.startswith(CUSTOM_CONNECTOR_PREFIX): raise e ...

即：当 source 类型以自定义连接器前缀（CUSTOM_CONNECTOR_PREFIX）开头时，workflow 不再按内置源目录结构推断模块路径，而是直接以sourcePythonClass字段值作为模块路径进行import_from_module动态导入。内置连接器则会走import_source_class(service_type, source_type)的常规注册表路径。

连接函数侧：从类名推导模块路径

除了 Source 类本身，连接相关的辅助函数（如get_connection/test_connection）也会按自定义规则导入。在 utils/importer.py 的import_connection_fn中：

if connection.type.value.lower().startswith(CUSTOM_CONNECTOR_PREFIX): python_class_parts = connection.sourcePythonClass.rsplit(".", 1) python_module_path = ".".join(python_class_parts[:-1]) _connection_fn = import_from_module("{}.{}".format(python_module_path, function_name)) else: _connection_fn = import_from_module( "metadata.ingestion.source.{}.{}.connection.{}".format( service_type.name.lower(), connection_type.value.lower(), function_name, ) )

可以看到：对于自定义连接器，框架会取sourcePythonClass中最后一个点号之前的部分作为模块路径，然后在该模块中查找名为function_name（例如get_connection、test_connection）的函数。这意味着：

• 你的自定义类可以放在任意包路径下，只要sourcePythonClass写得正确；
• 如果希望在连接测试/建连阶段有自定义逻辑，可在同一模块中额外提供get_connection、test_connection等函数，框架会自动发现并调用。

Test Connection：为什么是禁用的

Custom Database 连接器在 OpenMetadata UI 中禁用"Test Connection"（测试连接）按钮，原因很直接：这是自定义实现，OpenMetadata 无法预知你的源如何连接、如何鉴权。

原文给出的推荐做法是：在 ingestion 流程的第一步自行完成到数据源的连接校验。具体来说有两种落地方式：

1. 在自定义 Source 类的初始化（__init__）或首次next_record调用前主动探测源的可达性、凭据有效性，失败时抛出明确异常，让 workflow 在元数据抽取一开始就失败并暴露错误；
2. 在自定义类所在模块中提供test_connection函数（如上文import_connection_fn所示框架会尝试导入该函数），让 OpenMetadata 的测试连接能力得以扩展——从源码结构看，SAS 这类基于 Custom Database 机制实现的源即会在初始化阶段调用self.test_connection()（见 sas/metadata.py）。

从 workflow 的角度看，连接测试失败会直接阻断后续的元数据抽取与处理流程，因此将"连接校验前置"既是安全兜底，也是排查问题最快的切入点。

一个完整的 Custom Database 工作流配置示例

将以上配置项组合进 ingestion 工作流（YAML 配置或 UI 表单），一个典型的 Custom Database 元数据抽取配置如下：

source: type: customdatabase serviceName: my_custom_db_service serviceConnection: config: type: CustomDatabase sourcePythonClass: "my_company.connectors.my_csv_connector.MyCSVSource" connectionOptions: business_unit: "finance" file_path: "/data/export/sales_2026.csv" tableFilterPattern: excludes: - "temp_.*" sourceConfig: config: type: DatabaseMetadata sink: type: metadata-rest config: {} workflowConfig: openMetadataServerConfig: hostPort: "http://localhost:8585/api" authProvider: openmetadata securityConfig: jwtToken: "<your-jwt-token>"

配置要点回顾：

• type固定为customdatabase（连接内部类型为CustomDatabase）；
• sourcePythonClass填自定义类完整路径，类必须实现next_record；
• connectionOptions以键值对形式传入任意业务参数，在 Source 中通过self.service_connection.connectionOptions.__root__.get("key")读取；
• 可通过三个 filterPattern 字段做 schema/table/database 级过滤。

小结

Custom Database 是 OpenMetadata 面向"长尾数据源"设计的开放扩展点：它不限制你接入什么源，只约定两件事——提供一个实现了next_record的 Python 类，以及把连接参数通过connectionOptions传进去。配合 workflow 的动态导入机制（ingestion.py）与自定义连接函数发现机制（importer.py），即使 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