DataHub Metadata File 接入指南file Source 的配置、文件格式与状态化删除检测详解【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub导读Metadata File元数据文件是 DataHub 元数据摄取体系中的核心回放机制它可以作为存储与湖仓场景下元数据的载体也可以作为通用的元数据交换文件用于解耦元数据采集与元数据入库。本文基于 DataHub 官方文档 metadata-file 源集成说明结合仓库源码深入讲解file类型 Source 的完整配置、MCE/MCP 文件格式、读取模式、连接测试与状态化删除检测帮助读者在acryl-datahub环境中独立完成元数据文件生成 → 回放 → 校验的端到端工作流。一、Metadata File 集成概览根据官方文档Metadata File 是一个存储与湖仓storage and lakehouse平台而 DataHub 与 Metadata File 的集成覆盖了文件/湖仓元数据实体例如数据集datasets、路径paths与容器containers并且支持状态化删除检测stateful deletion detection。在仓库实现层面这一集成对应metadata-ingestion中的通用文件回放 Source——type: file。其源码位于 metadata-ingestion/src/datahub/ingestion/source/file.py类名GenericFileSource其类注释明确说明This plugin pulls metadata from a previously generated file. The metadata file sink can produce such files, and a number of samples are included in the examples/mce_files directory.也就是说该模块用于将先前生成好的元数据文件MCE / MCP JSON回放到 DataHub既可作为生产环境摄取工作流的一环也是调试元数据管线的利器。配合 Metadata File Sink 使用时可以形成从任意数据源采集 → 落盘为元数据文件 → 再回放进入 DataHub的解耦架构。二、概念映射Source 概念与 DataHub 概念官方文档给出了集成层面向的通用概念映射关系特定场景下的精确映射仍在完善中完整继承如下Source 概念DataHub 概念说明Platform/account/project scopePlatform Instance, Container在平台上下文内组织资产。Core technical asset如表/视图/主题/文件Dataset主要摄入的技术资产。Schema fields / columnsSchemaField在支持 Schema 抽取时包含。Ownership and collaboration principalsCorpUser, CorpGroup由支持所有权与身份元数据的模块发出。Dependencies and processing relationshipsLineage edges在支持且启用血缘抽取时可用。这张表的核心含义是无论元数据来源于哪个平台DataHub 最终都统一收敛到 Dataset、SchemaField、CorpUser/CorpGroup、Container、Platform Instance 等标准实体与关系上从而保证不同数据源在 DataHub 中的一致视图。例如仓库自带的示例文件 single_mce.json 中就包含一个CorpUserSnapshot其urn为urn:li:corpuser:harshal携带CorpUserInfo方面数据正是所有权主体 → CorpUser映射的具体落地。三、快速开始最小可用 Recipe官方在 file_recipe.yml 中给出了最小配置骨架source: type: file config: # Coordinates path: ./path/to/mce/file.json sink: # sink configs其中type: file指定使用GenericFileSourcepath指向待回放的元数据文件。若要组成完整的采集 → 落盘 → 回放链路可参照 Metadata File Sink 的 Quickstart recipe由任意 Source 采集后写入文件source: # source configs sink: type: file config: filename: ./path/to/mce/file.json注意Sink 侧早期配置字段为filename而 Source 侧的filename已被标记为废弃deprecated统一使用path源码中通过pydantic_renamed_field(filename, path, print_warningFalse)做了自动兼容迁移见 file.py。四、核心配置项详解fileSource 的全部配置集中在FileSourceConfig继承自StatefulIngestionConfigBase定义于 metadata-ingestion/src/datahub/ingestion/source/file.py。各字段含义如下字段必填默认值说明path✅无待摄入的文件或目录路径也可以是远程文件 URL。若指向目录则处理该目录下所有扩展名匹配file_extension默认.json的文件。file_extension否.json指向目录时用于过滤待处理文件的扩展名特殊值*表示处理目录下所有文件不区分扩展名。配置时会自动补全前导点add_leading_dot_to_extension校验器。read_mode否AUTO文件读取模式枚举值为STREAM/BATCH/AUTO详见第五节。aspect否无若设置只读取该 aspect 对应的元数据例如只回放ownership方面用于定向回放。count_all_before_starting否true启用时在开始前先统计文件总记录数用于精确估算完成时间若启动阶段耗时过长可关闭。stateful_ingestion否无状态化摄取配置类型为StatefulStaleMetadataRemovalConfig用于启用删除检测详见第七节。filename废弃否无已废弃自动迁移为path。从源码实现看path的解析并不局限于本地文件系统get_filenames()会先通过get_path_schema(path_str)解析路径协议再从fs_registry获取对应的文件系统实现见 file.py。这意味着只要注册表中存在对应协议即可直接摄入远程文件如对象存储等这是该模块面向存储/湖仓场景的重要支撑。aspect过滤逻辑在get_workunits_internal()中实现当对象为 MCP/MCPW 且aspectName与配置不一致时直接跳过见 file.py适合只回放某类方面的增量场景。五、支持的文件格式MCE 与 MCPGenericFileSource读取的是 DataHub 标准的元数据事件 JSON 文件支持两种主流对象判定逻辑位于_from_obj_for_file()见 file.pyMCEMetadataChangeEventJSON 对象中包含proposedSnapshot键对应实体快照例如single_mce.json中的CorpUserSnapshotMCP / MCPWMetadataChangeProposal / WrapperJSON 对象中包含aspect键对应单方面更新UsageAggregationClass包含bucket键属于已废弃的用量聚合格式读取到时会打印 warning 并丢弃Dropping deprecated UsageAggregationClass。每个对象读取后会执行item.validate()校验校验失败会抛出ValueError(fFailed to parse: {obj})。文件内部结构支持两种组织方式_iterate_file_batch中的兼容逻辑见 file.pyJSON 数组顶层为对象列表逐条回放如 mce_list.json 中包含两个CorpUserSnapshot单个 JSON 对象兼容旧版单对象格式直接回放。仓库 examples/mce_files 目录提供了大量可直接试用的样例除用户快照外还包括容器test_containers.json、域test_domains.json、结构化属性test_structured_properties.json、标签策略associate_tags_policies_test.json等是验证fileSource 行为的最佳素材。六、读取模式与性能AUTO / STREAM / BATCHread_mode控制文件的读取策略源码中枚举定义于FileReadModefile.py模式行为BATCH一次性json.load整个文件后逐条 yield适合中小文件。STREAM使用ijson增量解析流式读取内存占用低适合超大文件。AUTO自动决策文件大小小于 100MB_minsize_for_streaming_mode_in_bytes 100 * 1000 * 1000见 file.py时走 BATCH否则走 STREAM。决策逻辑位于_iterate_file()file.pyAUTO 模式按文件字节数切换模式STREAM 模式读取顶层item。流式模式下若开启count_all_before_starting会先完整扫描一遍统计元素总数用于进度百分比随后fp.seek(0)回到文件头开始正式读取file.py。配合这一机制FileSourceReport继承自StaleEntityRemovalSourceReport会记录每个文件的字节数、元素数、已读字节/元素数、解析/计数/反序列化耗时并实时计算完成百分比与预计剩余时间compute_stats见 file.py。在大文件批量回放场景下这些统计信息是监控进度与定位性能瓶颈的直接依据。七、状态化摄取与删除检测官方文档明确该集成支持状态化删除检测。在实现上GenericFileSource继承自StatefulIngestionSourceBase其允许启用的 Workunit 处理器为file.pyAutoWorkunitsReporterProcessor自动汇报 Workunit 进度AutoStaleEntityRemovalProcessor自动陈旧实体删除处理器即删除检测的核心当配置了stateful_ingestion时还会插入AutoStatusAspectProcessor用于维护实体的 status 方面。使用方式是在 recipe 的source.config下增加source: type: file config: path: ./path/to/mce/file.json stateful_ingestion: remove_stale_metadata: true启用后每次回放会以本次文件内容为基准对比状态化检查点将上次存在但本次文件中已不存在的实体识别为陈旧并触发删除从而保证 DataHub 中的元数据与源文件保持一致避免残留过时资产。这是元数据文件回放链路中维持数据一致性的关键能力与第五节中的aspect定向回放组合使用可实现细粒度的增量维护。八、连接测试Test ConnectionfileSource 声明了SourceCapability.TEST_CONNECTION能力且默认启用capability(SourceCapability.TEST_CONNECTION, Enabled by default)支持状态为 GAsupport_status(SupportStatus.GA)。其实现为静态方法test_connectionfile.py校验逻辑包括路径是否存在不存在则报告doesnt appear to be a valid file or directory是否为目录且是否可读os.access(path, os.R_OK)若为目录还需具备执行权限os.X_OK。因此在运行摄取前可以借助datahub ingest的--test-connection或 UI 中的测试连接功能快速验证path的可读性这与官方前置条件文档 file_pre.md 中确保对源元数据 API 具备读取权限的要求一致。九、故障排查官方文档 file_post.md 给出了通用的排查顺序先校验凭据、权限、连通性与范围过滤再查看摄取日志中的源相关错误并调整配置。结合源码fileSource 最常见的失败点与排查方向如下反序列化失败iterate_generic_file捕获每条记录的反序列化异常通过self.report.failure(messageFailed to deserialize metadata, contextf{file_status.path}-{i}, ...)记录具体文件与记录序号file.py排查时应优先检查该文件的 JSON 结构是否符合 MCE/MCP 格式缺少proposedSnapshot/aspect键会直接抛Unknown object type路径/权限问题参考第八节test_connection会给出具体的不可读、无执行权限提示启动过慢count_all_before_starting对超大文件会多一次全量扫描若启动时间不可接受可将其设为false扩展名未匹配指向目录时确保文件扩展名与file_extension默认.json一致或显式设为*。十、生成与回放构建解耦的元数据工作流综合上述内容fileSource 的典型生产用法是构建解耦式摄取管线任意数据源数据库、消息队列、湖仓等通过各自的 Source 采集元数据使用 Metadata File Sinksink.type: file将结果落盘为 MCE/MCP JSON 文件实现采集与入库的解耦也便于人工检查与调试中间产物在目标环境如生产 DataHub 实例使用type: file的 Source 回放该文件配合stateful_ingestion做删除检测通过aspect定向回放实现精细化更新借助test_connection与FileSourceReport的进度统计验证任务健康度。该模式在跨网络隔离环境迁移元数据、批量重放历史元数据、以及 CI 中做摄取回归测试等场景下尤为实用。仓库中 examples/mce_files 下的样例文件即可作为回放练习与单元验证的直接素材。延伸阅读源集成文档目录metadata-ingestion/docs/sources/metadata-file含 README.md、file_pre.md、file_post.md 与 file_recipe.ymlSource 核心实现metadata-ingestion/src/datahub/ingestion/source/file.py配套 Sink 文档metadata-ingestion/sink_docs/metadata-file.md示例元数据文件metadata-ingestion/examples/mce_files【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考