DataHub MongoDB 连接器实战DocumentDB 平台映射、权限与认证配置指南【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub导读本指南聚焦 DataHub 元数据采集框架中的 MongoDB 连接器source.type: mongodb系统讲解其核心能力边界、权限与认证配置、模式推断行为以及将 AWS DocumentDB 作为独立数据平台呈现的配置方法。文中配置与结论均以当前仓库metadata-ingestion模块的实际源码、官方文档与测试用例为据读完你将能写出可复制的生产级 MongoDB/DocumentDB 采集 Recipe并理解其底层实现原理。一、连接器能力总览MongoDB 连接器从 MongoDB 集群中采集元数据覆盖 DataHub 的以下核心实体见 mongodb 目录下的 README源概念DataHub 概念说明平台 / 账号 / 项目作用域Platform Instance、Container在平台上下文中组织资产核心技术资产表 / 视图 / 主题 / 文件Dataset主要采集的技术资产Schema 字段 / 列SchemaField支持 Schema 提取时包含所有权与协作者CorpUser、CorpGroup支持所有权与身份元数据的模块发出依赖与处理关系Lineage 边支持并启用血缘提取时可用从 mongodb.py 源码 的装饰器声明可以看到该模块的能力标注PLATFORM_INSTANCE默认启用通过platform_instance配置SCHEMA_METADATA默认启用通过enableSchemaInference控制CONTAINERS默认启用子类型为DATABASE即每个数据库生成一个 Database 容器支持度状态为GASupportStatus.GA。各能力是否需要在 Recipe 中额外配置以文档页顶部 Important Capabilities 表格为最终事实来源即 mongodb_post.md 中所述。此外连接器通过 MongoDB 兼容 API 同样支持 Amazon DocumentDB 的采集见 README.md并内置了状态化删除检测stateful deletion detection能力。二、前置条件权限与认证2.1 必需的数据库权限采集用户必须在每个待采集数据库上拥有read角色或者通过readAnyDatabase读取所有可访问数据库见 mongodb_pre.md// 授予指定数据库的读权限 db.grantRolesToUser(datahub_ingest, [{ role: read, db: your_database }]); // 或授予所有数据库的读权限 db.grantRolesToUser(datahub_ingest, [ { role: readAnyDatabase, db: admin }, ]);2.2 系统集合system.*的默认排除行为MongoDB 内部数据库admin、config、local在源码中被硬编码跳过见 mongodb.py 的DENY_DATABASE_LIST。同时连接器通过excludeSystemCollections选项默认开启排除system.profile、system.views等系统集合read角色足以覆盖全部标准采集场景若将excludeSystemCollections设为False采集system.profile需要dbAdmin角色从默认采集系统集合的旧版本升级、且 MongoDB 开启了 profiling 时可能遇到授权错误。此时可在 Recipe 中显式拒绝系统集合source: type: mongodb config: collection_pattern: deny: - .*\\.system\\..*源码中的处理逻辑与测试佐证在 get_workunits_internal 中当excludeSystemCollectionsTrue且集合名以system.开头时直接跳过并记入报告单元测试 test_mongodb_system_collections_excluded_by_default 验证了默认配置下只有users、orders等用户集合被采集而system.profile、system.views、system.indexes全部进入filtered报告test_mongodb_system_collections_included_when_opted_in 则验证了显式开启excludeSystemCollectionsFalse后system.views会被正常采集。2.3 认证机制authMechanismauthMechanism配置字段直接映射 PyMongo 的认证机制支持值如下见 mongodb_pre.md 与 mongodb.py 中的字段定义authMechanism适用场景必填字段DEFAULTSCRAM 认证MongoDB 默认根据服务端版本自动协商 SCRAM-SHA-256 或 SCRAM-SHA-1username、passwordSCRAM-SHA-256显式指定 SCRAM-SHA-256MongoDB 4.0username、passwordSCRAM-SHA-1显式指定 SCRAM-SHA-1username、passwordMONGODB-AWSMongoDB Atlas 或 AWS DocumentDB 的 AWS IAM 认证凭据通过boto3从环境中解析见下文MONGODB-X509X.509 证书认证通过connect_uri或options配置 TLS 选项用户名 / 密码DEFAULT、SCRAM最简单的配置直接提供username与passwordsource: type: mongodb config: connect_uri: mongodb://host:27017 username: ${MONGODB_USER} password: ${MONGODB_PASSWORD} authMechanism: DEFAULT从源码看username、password、authMechanism会被组装为pymongo.MongoClient的 options 传入密码使用TransparentSecretStr类型保护并在连接建立后执行admin.command(ping)做廉价连通性校验见 mongodb.py。password支持${VAR}环境变量注入示例 Recipe 见 mongodb_recipe.yml。AWS IAM 认证MONGODB-AWS设置authMechanism: MONGODB-AWS后连接器通过boto3的标准凭据链自动解析凭据环境变量AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_SESSION_TOKEN共享凭据 / 配置文件~/.aws/credentialsEC2 实例元数据 / 实例配置文件Instance ProfileECS 容器凭据EKS Pod Identity / IRSAIAM Roles for Service Accountssource: type: mongodb config: connect_uri: mongodbsrv://cluster.example.mongodb.net authMechanism: MONGODB-AWS hostingEnvironment: ATLAS # 或 AWS_DOCUMENTDB如需显式提供 AWS 凭据而非使用凭据链可将AWS_ACCESS_KEY_ID与AWS_SECRET_ACCESS_KEY直接填入username与passwordsource: type: mongodb config: connect_uri: mongodb://docdb-cluster.region.docdb.amazonaws.com:27017/?tlstruetlsCAFile/path/to/rds-combined-ca-bundle.pemreplicaSetrs0readPreferencesecondaryPreferredretryWritesfalse username: ${AWS_ACCESS_KEY_ID} password: ${AWS_SECRET_ACCESS_KEY} authMechanism: MONGODB-AWS hostingEnvironment: AWS_DOCUMENTDB platform: documentdb注意connect_uri中显式配置了 DocumentDB 的 TLS 参数tlstrue、tlsCAFile指向rds-combined-ca-bundle.pem、副本集与readPreferencesecondaryPreferred这些都是连接 AWS DocumentDB 的典型要求。三、以 DocumentDB 平台身份输出实体默认情况下无论底层源是 MongoDB 还是 AWS DocumentDB连接器都将所有实体输出在mongodb数据平台下。若希望将 DocumentDB 集群作为独立平台呈现需设置platform: documentdb见 mongodb_post.mdsource: type: mongodb config: connect_uri: ... authMechanism: MONGODB-AWS hostingEnvironment: AWS_DOCUMENTDB platform: documentdb3.1 配置校验规则该配置存在强约束platform: documentdb要求hostingEnvironment必须为AWS_DOCUMENTDB任何其他托管环境会在配置校验阶段被拒绝。这一逻辑由 MongoDBConfig 的模型校验器 实现model_validator(modeafter) def check_documentdb_requires_aws_hosting(self) - MongoDBConfig: if ( self.platform documentdb and self.hostingEnvironment ! HostingEnvironment.AWS_DOCUMENTDB ): raise ValueError( platformdocumentdb requires hostingEnvironmentAWS_DOCUMENTDB. ) return self托管环境枚举见 mongodb.py仅支持三个取值SELF_HOSTED默认、ATLAS、AWS_DOCUMENTDB。对应单元测试 test_platform_documentdb_without_aws_hosting_rejected_at_parse_time 验证了错误配置在解析阶段即抛错。3.2 平台切换对 URN 的影响将已有 Recipe 切换到platform: documentdb会生成新的documentdb数据集与容器 URN此前输出的mongodbURN 需要借助状态化采集stateful ingestion清理或手动软删除soft-delete。源码层面platform值贯穿所有实体构建路径数据集 URN 通过DatasetUrn.create_from_ids(platform_idself.platform, ...)生成见 mongodb.py容器键DatabaseKey(database..., platformself.platform, ...)与SchemaMetadata.platform同样携带该值见 mongodb.py 与 mongodb.py若配置了platform_instance还会生成携带 documentdb 平台的DataPlatformInstanceClass见 mongodb.py。测试 test_platform_documentdb_with_aws_hosting_uses_documentdb_platform 断言数据集 URN 为urn:li:dataset:(urn:li:dataPlatform:documentdb,mydb.users,PROD)且SchemaMetadata.platform为urn:li:dataPlatform:documentdbtest_platform_documentdb_with_platform_instance_propagates_to_all_urns 进一步验证 platform instance 段如prod-docdb会贯穿数据集、容器与 DataPlatformInstance 三类实体。而 test_aws_documentdb_hosting_without_platform_override_stays_mongodb 则确认不显式设置platform: documentdb时即使托管环境是 AWS DocumentDB实体仍输出在mongodb平台下以保证向后兼容。四、Schema 推断采样、类型映射与截断Schema 推断默认开启enableSchemaInference: True通过聚合管道对集合文档采样后推断字段类型。关键配置项见 mongodb_recipe.yml 与 mongodb.py配置项默认值说明enableSchemaInferenceTrue是否推断 SchemaschemaSamplingSize1000用于推断 Schema 的文档数设为null时扫描全部文档useRandomSamplingTrue是否随机采样False时从文档开头顺序选取maxSchemaSize300Schema 中最多包含的字段数maxDocumentSize16793600参与 Schema 推断的最大文档字节数MongoDB 单文档上限 16MB校验要求0 maxDocumentSize 167936004.1 采样与过滤的底层实现Schema 推断核心在 construct_schema_pymongo采样$sample随机采样或$limit顺序采样优先执行使后续聚合处理更小的数据集以提升性能文档大小过滤对服务器版本 ≥ 4.4且非 AWS DocumentDB的 MongoDB追加$addFields$match临时字段$bsonSize: $$ROOT过滤超限文档。是否启用由 should_add_document_size_filter 决定——因为$bsonSize操作在 4.4 之前不可用且 AWS DocumentDB 不支持。字段类型映射使用两张表见 mongodb.pyPYMONGO_TYPE_TO_MONGO_TYPE保留原生类型名如ARRAY、OBJECT、boolean、integer、biginteger、date、timestamp、oid、numberDecimal、binary、uuid、regex、javascript、minKey、maxKey、mixed等_field_type_mapping映射到 DataHub SchemaFieldDataType如ArrayTypeClass、BooleanTypeClass、StringTypeClass、TimeTypeClass、BytesTypeClass、RecordTypeClass、UnionTypeClass等。4.2 Schema 截断与降采样当推断出的字段数超过maxSchemaSize时连接器会按“字段出现频次降序、名称升序”排序后截断见 mongodb.py并在dataset_properties.customProperties中写入schema.downsampled: True与schema.totalFields: 实际字段数方便用户识别正在查看的是降采样后的 Schema。测试目录中亦有对应黄金文件佐证mongodb_mces_golden.json、mongodb_mces_no_random_sampling_golden.json、mongodb_mces_small_schema_size_golden.json。五、采集流程与过滤规则从源码的 get_workunits_internal 可以看出完整采集链路list_database_names()列出全部数据库跳过admin、config、local按database_patternAllowDenyPattern过滤数据库被过滤项记入report.filtered为每个数据库生成 Database 类型容器gen_containerslist_collection_names()列出集合跳过system.*默认并按collection_pattern过滤匹配名称为数据库.集合形式为每个集合构造 Dataset URN若开启 Schema 推断则生成SchemaMetadata并组装DatasetProperties、DataPlatformInstance等 aspect 输出 MCP WorkUnit。database_pattern与collection_pattern均支持allow/deny正则列表且数据库遍历与集合遍历均按排序顺序执行保证输出一致性。此外该 Source 继承StatefulIngestionSourceBase支持状态化采集stateful_ingestion配置可用于 URn 变更后的陈旧实体清理。六、局限性与故障排查6.1 已知限制模块行为受源 API、权限及平台暴露的元数据约束部分功能为条件支持或不受支持请以能力说明capability notes为准见 mongodb_post.md索引信息尚未采集——源码中留有 TODO使用list_indexes()或index_information()获取索引信息见 mongodb.py$bsonSize过滤在 MongoDB 4.4 与 AWS DocumentDB 上不生效对应场景下跳过文档大小过滤切换到platform: documentdb后旧mongodbURN 需自行清理见上文第三节。6.2 故障排查建议如果采集失败建议按以下顺序排查见 mongodb_post.md凭据确认username/password/authMechanism与认证方式匹配SCRAM、IAM、X.509权限确认采集用户具备read或readAnyDatabase角色若采集系统集合还需dbAdmin连通性确认connect_uri可访问TLS 参数如 DocumentDB 的 CA 证书正确作用域过滤检查database_pattern/collection_pattern是否误过滤了目标库/集合采集日志查看 ingestion 日志中的源相关错误report.warning会记录未知字段类型、Schema 降采样、被过滤项等据此调整配置。七、完整 Recipe 参考综合上述要点一份面向生产环境的 MongoDB 采集 Recipe 可组织如下基线来自 mongodb_recipe.ymlsource: type: mongodb config: # 连接坐标 connect_uri: mongodb://localhost # 凭据详见连接器文档中全部支持的 authMechanism 值含 MONGODB-AWS username: ${MONGODB_USER} password: ${MONGODB_PASSWORD} authMechanism: DEFAULT # 托管环境SELF_HOSTED默认/ ATLAS / AWS_DOCUMENTDB hostingEnvironment: SELF_HOSTED # 平台mongodb默认/ documentdb需 AWS_DOCUMENTDB 托管环境 platform: mongodb # 过滤规则 database_pattern: allow: - .* deny: [] collection_pattern: allow: - .* deny: - .*\\.system\\..* # Schema 推断选项 enableSchemaInference: True schemaSamplingSize: 1000 useRandomSampling: True maxSchemaSize: 300 maxDocumentSize: 16793600 # 状态化采集用于陈旧实体清理 stateful_ingestion: enabled: True remove_stale_metadata: True sink: # sink 配置 type: datahub-rest config: server: http://localhost:8080连接器完整源码位于 mongodb.py单元测试见 test_mongodb_source.py集成测试与黄金文件位于 tests/integration/mongodb/含 docker-compose.yml 与 test_mongodb.py文档入口见 mongodb 文档目录可按需深入研读。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考