背景与目标

背景

OpenMetadata 需要采集 DolphinScheduler的项目、工作流元数据，使使用者能在 OM 中查看 DS 项目列表、工作流定义、任务结构及运行状态。本方案采用 Java 实现。复用 OpenMetadata 已有的 Java ingestion 框架，不引入 Python 依赖。

设计目标

• 在 Java ingestion 框架中新增 Pipeline 类 Source。支持 DS 项目和工作流元数据采集
• 采集 DS 项目→ OM Project 实体
• 采集 DS 工作流定义→ OM Pipeline 实体
• 采集 DS 工作流运行实例→ OM PipelineStatus
• 复用已有的 DSApiHelper/DolphinSchedulerRESTClient HTTP 客户端基础设施
• 保持与现有 DatabaseRecord/TableRecord 一致的 Record 模型与 Sink 分发模式

使用者痛点概览

• No native Java source for DolphinScheduler: 当前只能通过 Python 脚本抓取，导致维护成本高。
• Poor error isolation: 单个项目或流程失败会影响整体采集流程。
• Sparse status history: 需要限制历史实例数量，否则对 OM 数据库运行速度造成压力。
• Difficult mapping of task dependencies and SQL extraction: Shell 任务常嵌入 SQL，手工解析繁琐。 怎么说呢，
• Lack of token fallback & SSL verification flexibility: 在生产环境中常需自定义认证方式。
• Pagination performance hit on large datasets:

总体架构

Java Ingestion 框架示意图

┌───────────────────────────────────────────────────────┐
│ JavaIngestionWorkflow │
│ prepare → testConnection → run → sink.write → close │
└───────────────────────────────────────────────────────┘
│ │
▼ ▼
┌────────────────────────────┐ ┌────────────────────────────┐
│ MetadataSource │ │ RestMetadataSink │
│ DolphinSchedulerPipeline │ │ │
│ Source │ ├─ ProjectRecord │
│ - listProjects │ ├─ PipelineRecord │
│ - listProcessDefinitions │ ├─ PipelineStatusRecord │
│ - getProcessDetail │ └─ …│
└────────────────────────────┘
▲ ▼
| ┌───────────────────────────────┐
| │ OpenMetadataClient SDK │
| └───────────────▲───────────────┘
| |
| |
▼ ▼

三层职责划分说明

层级	职责描述	关键类/接口名称
#Source 层	调用 DS REST API。解析响应，生成 Record 对象。隔离的观点是，单个请求失败不会终止整个流程。支持过滤模式与分页控制，

至于关键类。DolphinSchedulerPipelineSource,DSPipelineClient,DSHybridConnection,DSHybridModels...

#Record 模型层： 中立 DTO 用于 Source 与 Sink 的桥接，保持类型安全。怎么说呢，关键类这方面，ProjectRecord,PipelineRecord。PipelineStatusRecord,PipelineTaskInfo,TaskStatusInfo…

#Sink 层：按 Record 子类型分发到 OM SDK，实现持久化。再看关键类，RestMetadataSink .