DataHub MongoDB 连接器实战：DocumentDB 平台映射、权限与认证配置指南

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）：

从 mongodb.py 源码 的装饰器声明可以看到该模块的能力标注：

• PLATFORM_INSTANCE：默认启用（通过platform_instance配置）；
• SCHEMA_METADATA：默认启用（通过enableSchemaInference控制）；
• CONTAINERS：默认启用，子类型为DATABASE（即每个数据库生成一个 Database 容器）；
• 支持度状态为GA（SupportStatus.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 中，当excludeSystemCollections=True且集合名以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 则验证了显式开启excludeSystemCollections=False后system.views会被正常采集。

2.3 认证机制（authMechanism）

authMechanism配置字段直接映射 PyMongo 的认证机制，支持值如下（见 mongodb_pre.md 与 mongodb.py 中的字段定义）：

用户名 / 密码（DEFAULT、SCRAM）

最简单的配置，直接提供username与password：

source: 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的标准凭据链自动解析凭据：

1. 环境变量（AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_SESSION_TOKEN）
2. 共享凭据 / 配置文件（~/.aws/credentials）
3. EC2 实例元数据 / 实例配置文件（Instance Profile）
4. ECS 容器凭据
5. EKS Pod Identity / IRSA（IAM Roles for Service Accounts）

source: type: mongodb config: connect_uri: "mongodb+srv://cluster.example.mongodb.net" authMechanism: "MONGODB-AWS" hostingEnvironment: "ATLAS" # 或 "AWS_DOCUMENTDB"

如需显式提供 AWS 凭据而非使用凭据链，可将AWS_ACCESS_KEY_ID与AWS_SECRET_ACCESS_KEY直接填入username与password：

source: type: mongodb config: connect_uri: "mongodb://docdb-cluster.region.docdb.amazonaws.com:27017/?tls=true&tlsCAFile=/path/to/rds-combined-ca-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false" username: "${AWS_ACCESS_KEY_ID}" password: "${AWS_SECRET_ACCESS_KEY}" authMechanism: "MONGODB-AWS" hostingEnvironment: "AWS_DOCUMENTDB" platform: documentdb

注意：connect_uri中显式配置了 DocumentDB 的 TLS 参数（tls=true、tlsCAFile指向rds-combined-ca-bundle.pem）、副本集与readPreference=secondaryPreferred，这些都是连接 AWS DocumentDB 的典型要求。

三、以 DocumentDB 平台身份输出实体

默认情况下，无论底层源是 MongoDB 还是 AWS DocumentDB，连接器都将所有实体输出在mongodb数据平台下。若希望将 DocumentDB 集群作为独立平台呈现，需设置platform: documentdb（见 mongodb_post.md）：

source: type: mongodb config: connect_uri: "..." authMechanism: "MONGODB-AWS" hostingEnvironment: "AWS_DOCUMENTDB" platform: documentdb

3.1 配置校验规则

该配置存在强约束：platform: documentdb要求hostingEnvironment必须为AWS_DOCUMENTDB，任何其他托管环境会在配置校验阶段被拒绝。这一逻辑由 MongoDBConfig 的模型校验器 实现：

@model_validator(mode="after") def check_documentdb_requires_aws_hosting(self) -> "MongoDBConfig": if ( self.platform == "documentdb" and self.hostingEnvironment != HostingEnvironment.AWS_DOCUMENTDB ): raise ValueError( "platform='documentdb' requires hostingEnvironment='AWS_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_id=self.platform, ...)生成（见 mongodb.py）；
• 容器键DatabaseKey(database=..., platform=self.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:documentdb；test_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）：

4.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.py）：

• PYMONGO_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 可以看出完整采集链路：

1. list_database_names()列出全部数据库，跳过admin、config、local；
2. 按database_pattern（AllowDenyPattern）过滤数据库，被过滤项记入report.filtered；
3. 为每个数据库生成 Database 类型容器（gen_containers）；
4. list_collection_names()列出集合，跳过system.*（默认）并按collection_pattern过滤（匹配名称为数据库.集合形式）；
5. 为每个集合构造 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）：

1. 凭据：确认username/password/authMechanism与认证方式匹配（SCRAM、IAM、X.509）；
2. 权限：确认采集用户具备read或readAnyDatabase角色；若采集系统集合还需dbAdmin；
3. 连通性：确认connect_uri可访问，TLS 参数（如 DocumentDB 的 CA 证书）正确；
4. 作用域过滤：检查database_pattern/collection_pattern是否误过滤了目标库/集合；
5. 采集日志：查看 ingestion 日志中的源相关错误（report.warning会记录未知字段类型、Schema 降采样、被过滤项等），据此调整配置。

七、完整 Recipe 参考

综合上述要点，一份面向生产环境的 MongoDB 采集 Recipe 可组织如下（基线来自 mongodb_recipe.yml）：

source: 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