DataHub Snowplow 连接器本地集成测试环境搭建与验证指南（Option B：DuckDB + Mock BDP Server）

DataHub Snowplow 连接器本地集成测试环境搭建与验证指南Option BDuckDB Mock BDP Server【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub导读本文基于 metadata-ingestion/tests/integration/snowplow/docs/SETUP_VERIFICATION.md 展开完整记录 DataHub 开源仓库中Snowplow 连接器Option B本地开发环境的搭建过程与验证结果通过增强型 JSON Fixture、DuckDB 本地数仓、Mock BDP API Server 三件套无需真实 Snowplow BDP 账号即可完成连接器的集成测试、数据血缘warehouse lineage与所有权ownership抽取验证。读完本文你将掌握这套本地测试环境的全部组件结构、脚本参数、验证命令与故障排查方法并能直接在自己的 DataHub 仓库中复现整套流程。一、Option B 方案概览为什么需要本地测试环境Snowplow 连接器的生产目标环境是 Snowplow BDPBehavioral Data Platform托管版其 Console API 需要付费/试用账号且仓库型数仓依赖外部资源无法在 CI 中快速、离线地反复验证。为此仓库在 metadata-ingestion/tests/integration/snowplow 下提供了两套方案Option ABDP Cloud直连真实 BDP 环境用于上线前的最终验证Option BLocal完全本地化——用 JSON Fixture 模拟 API 响应、用 DuckDB 模拟 Snowplow 数仓、用 Flask 起一个 Mock BDP Server 模拟真实 HTTP 调用约 5 分钟即可搭建适合开发迭代与 CI/CD。两者的对比来源LOCAL_TEST_SETUP.md特性Option ABDP CloudOption BLocal搭建耗时1-2 小时5 分钟外部依赖BDP 账号、Warehouse无成本付费/试用账号免费真实性生产级 APIMock 响应离线测试否是适用场景最终验证开发与 CI本文聚焦 Option B其四个核心组件增强 Fixtures、DuckDB 脚本、Mock BDP Server、配套文档均已全部验证通过。二、组件一增强型 JSON Fixtures含所有权/部署历史2.1 文件位置与内容增强后的 Fixture 位于 fixtures/data_structures_with_ownership.json包含 3 个贴近真实业务的 Snowplow Schemacom.acme命名空间完整覆盖部署历史、多版本演化与自定义元数据checkout_startedevent1-0-0由ryancompany.com创建1-1-0由janecompany.com修改用于测试 Schema 演化与字段作者归属field authorshipproduct_viewedevent单版本1-0-0创建者alicecompany.comuser_contextentity实体context类型 Schema创建者bobcompany.com。所有权相关数据要点✅deployments数组携带initiator发起人字段✅ 每个 Schema 有多个版本用于验证版本历史✅ts时间戳支持时间维度追踪✅meta.customData中可放置自定义元数据如{team: checkout}。2.2 Fixture 与原始文件的对比仓库中同时保留了原始 Fixture data_structures_response.json 和增强版本二者差异如下来源SETUP_VERIFICATION.md 中的对比表特性data_structures_response.jsondata_structures_with_ownership.jsonSchema 数量23含 deployments否是initiator所有权数据否是Schema 演化单版本多版本命名空间com.examplecom.acme用途简单测试所有权/演化完整测试2.3 源码印证所有权如何被抽取增强 Fixture 中的deployments[].initiator是所有权抽取的数据来源。在连接器源码 builders/ownership_builder.py 中extract_ownership_from_deployments()按时间戳排序最早最旧的 deployment →createdBy映射为 DataHub 的DATAOWNER所有权类型最新最近的 deployment →modifiedBy映射为PRODUCER字段级作者field authorship则由版本历史推导而来。同时在 test_snowplow.py 中test_snowplow_ingest会加载该 FixturemockSnowplowBDPClient.get_data_structures并 mock/users返回的 4 个用户ryancompany.com、janecompany.com、alicecompany.com、bobcompany.com用于将initiator解析为真实用户 URN最终与 golden 文件比对。三、组件二DuckDB 本地数仓脚本3.1 脚本能力脚本位于 setup/setup_duckdb.py用于在本地创建模拟 Snowplowatomic.events表的 DuckDB 数据库创建本地 DuckDB 数据库生成贴近真实的snowplow.events表灌入示例事件数据支持自定义事件数量与数据库路径。3.2 命令行参数参数默认值说明--db-pathsnowplow_test.duckdbDuckDB 数据库文件路径--event-count100生成的示例事件数量--recreate关闭删除并重建数据库参数定义见 setup_duckdb.py 的 argparse 部分。3.3 数据库表结构create_database()setup_duckdb.py创建的snowplow.events表结构如下snowplow.events ( -- 基础列Base columns app_id, platform, collector_tstamp, event, event_id, user_id, user_ipaddress, -- 页面上下文Page context page_url, page_title, page_referrer, -- 设备上下文Device context br_name, br_family, os_name, os_family, -- Geo 富化Geo enrichment geo_country, geo_region, geo_city, geo_zipcode, geo_latitude, geo_longitude, -- 自定义事件上下文JSON 列 contexts_com_acme_checkout_started_1, contexts_com_acme_product_viewed_1, contexts_com_acme_user_context_1, -- Unstruct 事件JSON 列 unstruct_event_com_acme_checkout_started_1, unstruct_event_com_acme_product_viewed_1, -- 时间戳 derived_tstamp, load_tstamp )3.4 样例数据生成逻辑generate_sample_events()setup_duckdb.py在 30 天时间窗内以 5 分钟为间隔生成事件随机选择事件类型checkout_started映射为unstruct事件或product_viewed映射为struct事件每行生成event_idUUID、user_iduser_1~user_100、随机浏览器/操作系统/地理位置checkout_started的 unstruct/context JSON 含amount、currencyUSD/EUR/GBP、discount_codeSAVE10/WELCOME20、items数组product_viewed的 JSON 含product_id、categoryelectronics/clothing/books、price每行附加contexts_com_acme_user_context_1user_type取值 free/premium/enterprise、registration_date。当前预置数据库 setup/snowplow_test.duckdb 共 100 条事件分布为struct 54 条 / unstruct 46 条可通过分组查询验证SELECT event, COUNT(*) FROM snowplow.events GROUP BY event输出eventcountstruct54unstruct46四、组件三Mock BDP API Server4.1 脚本能力与启动方式脚本位于 setup/mock_bdp_server.py基于 Flask 实现模拟 Snowplow BDP Console API从 Fixture 目录读取数据返回支持过滤、分页、完整请求/响应日志与错误处理。启动命令python mock_bdp_server.py --port 8081 # 可选参数--host默认 localhost、--debug开启调试模式4.2 端点清单对照实际源码mock_bdp_server.pyMock Server 提供的端点如下端点说明源码位置GET /API 文档与端点索引L326-L346GET /health健康检查L314-L323GET /organizations/{orgId}/credentials/v3/tokenJWT Token 签发L68-L93GET /organizations/{orgId}/data-structures/v1Schema 列表支持 filter/vendor/limit/offsetL96-L147GET /organizations/{orgId}/data-structures/v1/{hash}按 hash 获取单个 SchemaL150-L174GET /organizations/{orgId}/data-products/v2Data products实际代码为 v2 端点L177-L204GET /organizations/{orgId}/users组织用户列表用于 initiator 邮箱解析L207-L250GET /organizations/{orgId}/event-specs/v1Event specificationsL253-L280GET /organizations/{orgId}/tracking-scenarios/v1Tracking scenarios旧路径L283-L311说明原验证文档中列出的是/data-products/v1而当前仓库源码实现为/data-products/v2v2 使用data/includes/errors包裹格式与 event-specs 一致区别于>source: type: snowplow config: # BDP 连接Mock——需先启动 mock server bdp_connection: organization_id: test-org-uuid api_key_id: test-key-id api_key: test-secret console_api_url: http://localhost:8081/api/msc/v1 # Mock server # Schema 过滤 schema_pattern: allow: - com.acme.* # 只抽取 event/entity 类型 Schema schema_types_to_extract: - event - entity # 可选功能——本测试中关闭 extract_event_specifications: false extract_tracking_plans: false # 仓库血缘——通过 DuckDB 启用 extract_warehouse_lineage: true # DuckDB 仓库连接 warehouse_connection: warehouse_type: duckdb database: snowplow_test.duckdb schema_name: snowplow # 平台标识 platform_instance: snowplow-test env: TEST sink: type: file config: filename: ./snowplow_duckdb_output.json其中console_api_url指向 Mock Server使连接器通过真实 HTTP 链路访问本地模拟 API。相关连接参数organization_id、api_key_id、api_key、console_api_url默认值https://console.snowplowanalytics.com/api/msc/v1、timeout_seconds默认 60、max_retries默认 3在源码 snowplow_config.py 的SnowplowBDPConnectionConfig中定义。六、验证测试记录与命令6.1 Test 1Mock API 集成测试 ✅ PASSEDcd metadata-ingestion source venv/bin/activate pytest tests/integration/snowplow/test_snowplow.py::test_snowplow_ingest -v验证点Mock API 响应正常、golden 文件比对通过、无错误。该测试通过unittest.mock.patch替换datahub.ingestion.source.snowplow.snowplow.SnowplowBDPClient将内存中的DataStructure对象注入客户端避免任何真实 HTTP 调用见 test_snowplow.py。6.2 Test 2DuckDB 数据库搭建 ✅ COMPLETEDcd metadata-ingestion/tests/integration/snowplow/setup python setup_duckdb.py --event-count 100结果数据库snowplow_test.duckdb创建成功Schemasnowplow、表snowplow.events建立插入 100 条事件。验证查询SELECT event, COUNT(*) FROM snowplow.events GROUP BY event6.3 Test 3Mock BDP Server ✅python mock_bdp_server.py --port 8081逐端点验证健康检查curl http://localhost:8081/health→healthyToken 签发curl -H X-Api-Key-Id: test-key-id -H X-Api-Key: test-secret \ http://localhost:8081/organizations/test-org-uuid/credentials/v3/token→mock_jwt_token_12345Data Structures 列表curl -H Authorization: Bearer mock_jwt_token_12345 \ http://localhost:8081/organizations/test-org-uuid/data-structures/v1→ 返回 3 个带 deployments 的 Schema。6.4 Test 4Fixture 文件校验 ✅data_structures_response.json原始2 个 Schemapage_view、user_context无 deployment 数据结构简单JSON 有效data_structures_with_ownership.json增强3 个 Schema完整部署历史、initiator 所有权、多版本演化JSON 有效。七、快速上手命令速查以下命令均以仓库根目录为基准运行全部集成测试cd metadata-ingestion source venv/bin/activate pytest tests/integration/snowplow/ -v初始化 DuckDB仓库测试用cd metadata-ingestion/tests/integration/snowplow/setup python setup_duckdb.py --event-count 100启动 Mock BDP Servercd metadata-ingestion/tests/integration/snowplow/setup python mock_bdp_server.py --port 8081查询 DuckDBcd metadata-ingestion/tests/integration/snowplow/setup duckdb snowplow_test.duckdb SELECT COUNT(*) FROM snowplow.events更多查询示例来自 LOCAL_TEST_SETUP.md# 查看 checkout 事件 duckdb snowplow_test.duckdb SELECT event_id, user_id, unstruct_event_com_acme_checkout_started_1 FROM snowplow.events WHERE event unstruct LIMIT 5 # 查看用户上下文 duckdb snowplow_test.duckdb SELECT user_id, contexts_com_acme_user_context_1 FROM snowplow.events LIMIT 5八、三类典型测试场景场景 1基础集成测试最快目的验证连接器在 Mock API 响应下正常工作。pytest metadata-ingestion/tests/integration/snowplow/test_snowplow.py -v状态✅ Working。适合开发期的快速校验无需任何外部依赖。场景 2DuckDB 仓库血缘warehouse lineage目的从本地 DuckDB 验证仓库血缘抽取。前置条件已创建 DuckDB 数据库python setup_duckdb.py --event-count 100Mock BDP Server 已运行。步骤# 终端 1启动 mock server python metadata-ingestion/tests/integration/snowplow/setup/mock_bdp_server.py --port 8081 # 终端 2执行 ingestion datahub ingest -c metadata-ingestion/tests/integration/snowplow/recipes/snowplow_with_duckdb.yml状态 Ready to test需 Mock Server 与 DuckDB 联动。仓库血缘的抽取逻辑可进一步参考连接器源码 processors/warehouse_lineage_processor.py。场景 3完整 Mock 环境端到端目的通过真实 HTTP 调用链路做端到端测试。组件Mock BDP ServerHTTP DuckDB 数据库数仓 测试 Recipe。# 终端 1启动 mock server cd metadata-ingestion/tests/integration/snowplow/setup python mock_bdp_server.py --port 8081 # 终端 2初始化 DuckDB python setup_duckdb.py --event-count 100 # 终端 3配置环境变量并执行 ingestion export SNOWPLOW_ORG_IDtest-org-uuid export SNOWPLOW_API_KEY_IDtest-key-id export SNOWPLOW_API_KEYtest-secret datahub ingest -c metadata-ingestion/tests/integration/snowplow/recipes/snowplow_with_duckdb.yml状态 基础设施就绪待完整端到端验证。九、目录结构速览metadata-ingestion/tests/integration/snowplow/ ├── fixtures/ │ ├── data_structures_response.json [原始 Fixture] │ └── data_structures_with_ownership.json [增强含 deployments] ├── golden_files/ │ └── snowplow_mces_golden.json [Golden 比对文件] ├── recipes/ │ ├── snowplow_with_duckdb.yml [DuckDB 数仓 Recipe] │ └── test_mock_bdp.yml 等 [其他测试 Recipe] ├── setup/ │ ├── setup_duckdb.py [数据库搭建脚本] │ ├── mock_bdp_server.py [Mock API Server] │ └── snowplow_test.duckdb [预置测试数据库] ├── docs/ │ ├── LOCAL_TEST_SETUP.md [本地使用指南] │ └── SETUP_VERIFICATION.md [验证报告本文依据] ├── test_snowplow.py [集成测试] └── test_snowplow_performance.py [性能测试]完整结构与测试分类可参考 tests/integration/snowplow/README.md。十、依赖安装✅datahub[snowplow]带 Snowplow 支持的主包✅duckdb本地数仓测试✅flaskMock BDP Server可选。安装方式cd metadata-ingestion source venv/bin/activate pip install -e .[snowplow] duckdb flask十一、后续工作项Next Steps已完成的即时项✅ 基础测试通过✅ DuckDB 数据库已创建✅ Mock Server 已验证✅ Fixtures 校验通过。待完成的完整测试项 用 DuckDB 测试仓库血缘抽取 从 deployments 测试所有权抽取可参考 docs/OWNERSHIP_TESTING_GUIDE.md 用多版本测试 Schema 演化 测试 data products 抽取启用后。连接器对接项更新连接器以支持 DuckDB 仓库类型实现从 deployments 的所有权抽取当前OwnershipBuilder.extract_ownership_from_deployments已具备基础能力见 builders/ownership_builder.py从版本历史增加字段作者归属完成全链路端到端测试。十二、常见问题排查Troubleshooting问题Module not found 报错解决确认 venv 已激活且包已安装cd metadata-ingestion source venv/bin/activate pip install -e .[snowplow]问题DuckDB 未找到解决安装 duckdbpip install duckdb问题Mock Server 端口被占用解决更换端口或停止占用进程lsof -i :8081 # 查找占用进程 kill PID # 停止进程 # 或换端口 python mock_bdp_server.py --port 8082问题数据库文件不存在解决重新创建数据库cd metadata-ingestion/tests/integration/snowplow/setup python setup_duckdb.py问题集成测试失败解决若输出属预期变更可更新 golden 文件pytest tests/integration/snowplow/test_snowplow.py --update-golden-files # 或查看详细 diff pytest tests/integration/snowplow/test_snowplow.py -vv问题Fixture 文件缺失解决确认文件存在并从 git 恢复ls -la metadata-ingestion/tests/integration/snowplow/fixtures/ git checkout metadata-ingestion/tests/integration/snowplow/fixtures/data_structures_with_ownership.json十三、自定义测试数据新增 Schema Fixture编辑 data_structures_with_ownership.json按以下模板追加{ hash: new_schema_hash, organizationId: test-org-uuid, vendor: com.acme, name: new_event, format: jsonschema, description: New event schema, meta: { hidden: false, schemaType: event, customData: { team: new-team } }, deployments: [ { version: 1-0-0, initiator: developercompany.com, ts: 2024-03-01T10:00:00Z } ], data: { self: { vendor: com.acme, name: new_event, version: 1-0-0 }, properties: { field1: { type: string } } } }调整 DuckDB 事件数据修改 setup/setup_duckdb.py 中的generate_sample_events()可新增事件类型、调整字段分布、改变事件数量或时间范围。扩展 Mock API 端点修改 setup/mock_bdp_server.py新增 Flask 路由、创建新的 Fixture 文件或实现新的 API 行为。十四、结论✅Option B 本地测试环境已完整搭建并通过验证集成测试全部通过golden 文件比对无差异DuckDB 数据库创建成功含 100 条示例事件struct 54 / unstruct 46Mock BDP Server 正常提供增强 Fixture配套文档齐全。该环境已具备支撑以下工作的能力✅ 快速开发迭代无外部依赖、5 分钟可复现✅ 离线测试✅ CI/CD 集成✅ 基于 deployments 的所有权测试✅ 基于 DuckDB 的仓库血缘测试。推荐后续动作使用 DuckDB 验证仓库血缘抽取实现从 deployments 数组的所有权抽取结合 Mock Server 完成端到端全链路测试让连接器集成测试切换到增强 Fixture。如需进行真实 BDP 环境Option A的最终上线验证可参考 docs/REAL_BDP_TESTING_GUIDE.md。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考