OpenMetadata Custom Dashboard 连接器实战：把任何自研报表系统接入元数据摄取流程

OpenMetadata Custom Dashboard 连接器实战把任何自研报表系统接入元数据摄取流程【免费下载链接】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 Dashboard 连接器文档CustomDashboard.md详解该连接器Source Python Class Connection Options两个核心配置项的设计意图与读取方式并结合仓库中可直接运行的参考实现 custom_dashboard.py、集成测试 test_custom_connectors.py 与 README.md给出一个从连接配置到端到端验证的完整落地方案读完后你能够编写自己的 Python 类将仅存在于内部工程环境中的报表/仪表盘系统注册为 OpenMetadata 的 Dashboard 服务源。Custom Dashboard 连接器是什么Custom Dashboard 是 OpenMetadata 摄取ingestion框架中一组自定义连接器之一。文档给出的定位非常明确它本质上是一个包装器wrapper用于包装你自行编写、并打入 OpenMetadata ingestion 镜像的任意 Python 类。设计目标是——当某个数据源只存在于你所在业务或工程团队的私有上下文中比如自研 BI 平台、内部报表系统、自研看板服务时你不需要等官方开发对应连接器而是可以把这套工具拿过来把该系统的元数据图表、仪表盘、数据模型等搬运进 OpenMetadata。Custom Dashboard 只是这一族连接器中的一个。从 README.md 的测试清单可以看到仓库按服务类型提供了多个同构的自定义连接器示例模块服务类型产出实体custom_database.pyCustomDatabase1 个 database → 2 个 schema → 3 张带列的表custom_dashboard.pyCustomDashboard2 个图表 1 个引用它们的仪表盘custom_messaging.pyCustomMessaging2 个带分区与消息 schema 的 topiccustom_mlmodel.pyCustomMlModel1 个带特征、超参数与存储位置的模型custom_pipeline.pyCustomPipeline1 条含 3 个串联任务的流水线custom_search.pyCustomSearch2 个带类型化字段的索引custom_storage.pyCustomStorage根容器 子容器 数据模型custom_drive.pyCustomDrive目录 文件、电子表格 工作表其中CustomDashboard类型对应 custom_dashboard.py它不连接任何外部系统仅从内存中产出确定性的仪表盘实体——这正是理解该连接器契约的最小样本。连接配置详解Custom Dashboard 的连接配置只有两个核心字段下面逐一展开并给出源码层面的印证。Source Python ClasssourcePythonClass这是摄取工作流Workflow在运行时实例化的Python 类名模块点分路径。也就是说OpenMetadata 并不内置这个类的逻辑而是按你填写的字符串反射加载、实例化你的类。文档对此的关键约束是该类需要实现记录迭代方法官方文档表述为next_record以便 Workflow 能够持续读取记录并向 OpenMetadata API 发送。结合当前仓库的实际实现可以更具体地描述这个契约。参考实现 CustomDashboardSource 继承自metadata.ingestion.api.steps.Source并按 README.md 中的说明实现create、prepare、test_connection、close和_iter方法其中迭代器负责把记录持续吐给 Workflowclass CustomDashboardSource(Source): Yields two charts, a data model and one dashboard referencing both. def __init__(self, config: WorkflowSource, metadata: OpenMetadata): super().__init__() self.config config self.metadata metadata self.service_connection config.serviceConnection.root.config classmethod def create( cls, config_dict: dict, metadata: OpenMetadata, pipeline_name: str | None None, ) - CustomDashboardSource: config: WorkflowSource WorkflowSource.model_validate(config_dict) connection config.serviceConnection.root.config if not isinstance(connection, CustomDashboardConnection): raise InvalidSourceException(fExpected CustomDashboardConnection, but got {connection}) return cls(config, metadata) def _iter(self, *_, **__) - Iterable[Either]: # 依次产出 CreateDashboardServiceRequest / CreateChartRequest / ... ...从源码结构看迭代产出的每条记录都是一个Either其right为对应的Create*Request对象且必须以 service 请求开头保证服务实体先于子实体存在。参考实现中的产出顺序是CreateDashboardServiceRequest→ 2 个CreateChartRequestrevenue_by_month折线图、orders_by_region柱状图→ 1 个CreateDashboardDataModelRequest名为my_revenue_model带day/revenue两列的语义模型→ 1 个CreateDashboardRequestmy_sales_overview通过charts与dataModels字段引用前述图表与模型def _iter(self, *_, **__) - Iterable[Either]: service_name self.config.serviceName yield Either( rightCreateDashboardServiceRequest( nameservice_name, serviceTypeDashboardServiceType.CustomDashboard, connectionself.config.serviceConnection.root, displayNameCustom Dashboard Demo, descriptionReporting served by the custom dashboard connector, ) ) for chart_name, chart_type, chart_description in CHARTS: yield Either( rightCreateChartRequest( namechart_name, displayNamechart_name.replace(_, ).title(), descriptionchart_description, chartTypechart_type, serviceservice_name, sourceUrlfhttps://dashboards.example.com/chart/{chart_name}, ) ) yield Either( rightCreateDashboardRequest( nameDASHBOARD_NAME, displayNameMy Sales Overview, descriptionDashboard produced by the custom dashboard connector, dashboardTypeDashboardType.Dashboard, serviceservice_name, charts[f{service_name}.{chart_name} for chart_name, _, _ in CHARTS], dataModels[f{service_name}.model.{DATA_MODEL}], sourceUrlfhttps://dashboards.example.com/dashboard/{DASHBOARD_NAME}, ) )其中有两个值得注意的实现细节图表与仪表盘的关联通过 FQN{service_name}.{chart_name}表达仪表盘对数据模型的引用则带model.前缀{service_name}.model.{data_model_name}custom_dashboard.py 的文件头注释明确提示DataModelType枚举目前没有厂商中性的取值自定义连接器只能借用一个既有取值示例中用的是DataModelType.SupersetDataModel。如果你的自研系统有语义层/数据集模型同样需要先选一个最接近的既有枚举值。Connection OptionsconnectionOptionsconnectionOptions用于把输入参数从摄取 YAML 传入你的 Source 类这是自定义连接器的参数化通道。文档给出的典型场景是你希望根据参数business_unit的值走不同的取数逻辑那么在连接配置里写入键business_unit及任意值再在 Source 类中这样读取business_unit self.service_connection.connectionOptions.__root__.get(business_unit)关于取值类型README.md 有一句重要的工程约束connectionOptions的值是Dict[str, str]数字型选项必须由连接器自行解析。也就是说如果你传limit: 100需要在类内做int(...)转换不要假设它已经是数值类型。完整落地流程一个可运行的 Custom Dashboard 摄取下面把参考实现串成一条可复现的实操路径。这些示例代码位于 ingestion/tests/integration/custom_connectors/ 目录不触达任何外部系统唯一要求是一台运行中的 OpenMetadata 服务器默认http://localhost:8585。第 1 步把你的连接器模块放到 PYTHONPATH 中自定义连接器模块就是普通的顶层 Python 模块不需要注册进任何插件系统。把连接器所在目录加入PYTHONPATH然后在 YAML 中用模块名.类名指向它即可。第 2 步编写摄取工作流 YAML以本仓库 dashboard 示例为蓝本对照 test_custom_connectors.py 中的ConnectorSpecsource_typecustom-dashboard、connection_typeCustomDashboard、source_python_classcustom_dashboard.CustomDashboardSource、source_config_typeDashboardMetadata一个可运行的工作流配置如下source: type: custom-dashboard serviceName: custom_dashboard_demo serviceConnection: config: type: CustomDashboard sourcePythonClass: custom_dashboard.CustomDashboardSource connectionOptions: business_unit: finance # 任意自定义键值对类型均为字符串 sourceConfig: config: type: DashboardMetadata sink: type: metadata-rest config: {} workflowConfig: openMetadataServerConfig: hostPort: http://localhost:8585/api authProvider: openmetadata securityConfig: jwtToken: ingestion-bot token各字段与文档对应关系type: CustomDashboard连接类型决定config.serviceConnection.root.config反序列化出的模型参考实现中即断言其为CustomDashboardConnection否则抛出InvalidSourceException见 create 方法sourcePythonClassWorkflow 实例化的类模块点分路径connectionOptions传入你类的任意字符串参数sourceConfig的type: DashboardMetadata声明按 Dashboard 实体族解析源配置。第 3 步运行摄取PYTHONPATHingestion/tests/integration/custom_connectors metadata ingest -c workflow.yaml第 4 步用集成测试验证产出仓库自带端到端验证source env/bin/activate cd ingestion python -m pytest tests/integration/custom_connectors -q对 dashboard 连接器test_custom_connectors.py 断言摄取后必须产出以下实体Chart:revenue_by_month、orders_by_regionDashboardDataModel:model.my_revenue_model且其列必须恰为day、revenue并均带 descriptionDashboard:my_sales_overview。如果你的自定义连接器跑通后实体缺失或 FQN 对不上通常问题就出在_iter的产出顺序service 必须最先或 FQN 拼接格式上。关于 Test Connection为什么该连接器禁用了连接测试文档在 Test Connection 一节说明由于这是自定义实现UI 上的连接测试被有意禁用推荐的替代做法是把验证与源系统的连接作为摄取流程的第一步来完成。这一点在参考实现中得到了直接印证——CustomDashboardSource.test_connection是一个空实现No external system to reach见 custom_dashboard.py。对真实自研系统的连接器建议把文档的建议落到代码里在你的test_connection或prepare中完成鉴权、连通性、表/看板可见性等检查失败即抛异常让摄取在最早期就暴露配置错误而不是等实体开始流转后才发现问题。编写自定义 Dashboard 连接器的要点清单综合文档与 README.md 中 Writing your own 一节编写一个新的 Custom Dashboard或其他自定义服务类型连接器需要满足继承metadata.ingestion.api.steps.Source实现create含连接类型校验、prepare、test_connection、close、_iter五个方法_iter产出Either(rightCreateXRequest)并以get_create_service_from_source(...)之类的 service 请求打头保证服务先于子实体连接参数走connectionOptions用self.service_connection.connectionOptions.__root__.get(key)读取并对数值/布尔参数自行做类型转换若改走*ServiceSource基类复用某个官方服务源的通用逻辑README 提示还需要在类所在模块额外提供模块级的get_connection与test_connection函数无厂商中性枚举值时借用最接近的既有枚举如示例借用SupersetDataModel作为数据模型类型并在注释中说明原因避免后来者误以为是笔误连接测试自行兜底UI 不提供一键测试把源系统验证前移到摄取流程首步。小结Custom Dashboard 连接器本身不取数它提供的是逃生舱式的扩展机制sourcePythonClass把实例化权交还给你的 Python 类connectionOptions提供参数化通道摄取框架负责后续的记录迭代与 API 写入。仓库内的 custom_dashboard.py 是一个 129 行的完整可运行范本覆盖了服务、图表、数据模型、仪表盘四类实体的产出与 FQN 关联配合 test_custom_connectors.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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考