DataHub dlt 连接器实战从 dlt 本地状态目录提取 Pipeline、DataJob 与血缘元数据【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub导读本文讲解 DataHub 内置的dltdata load tool摄取连接器它直接读取 dlt 写入本地的管线状态目录默认~/.dlt/pipelines/无需实时连接 dlt 或目标库即可提取 Pipeline 元数据并将 dlt 管线建模为 DataFlow、每个目标表建模为 DataJob同时输出出口Outlet与入口Inlet血缘、列级血缘以及可选的运行历史。读完本文你将掌握 dlt 连接器的数据模型、pipelines_dir在不同部署形态下的配置方式、血缘缝合的关键参数以及运行历史、状态化摄取等进阶能力的启用方法。一、连接器概览dlt 模块摄取什么dltdata load tool是面向数据加载的 Python 库DataHub 的 dlt 连接器从 dlt 的本地状态目录读取管线元数据并建模为 DataHub 实体dlt.py 是连接器主入口类DltSource标注平台名dlt、支持状态BETA。基本元数据提取不需要与 dlt 或目标数据源建立实时连接——连接器只读取本地文件。若摄取环境中安装了 dlt Python 包则优先走 dlt SDK 获取更丰富的元数据否则回退为直接解析 schema YAML 状态文件。连接器摄取的核心实体与血缘包括实体建模粒度说明DataFlow每个 dlt 管线一个pipeline_name通过_build_dataflow生成携带destination、dataset_name、pipelines_dir、working_dir等自定义属性subtype 为dlt Pipeline见 subtypes.pyDataJob每个目标表一个含自动展开的子表如orders__items通过_build_datajob生成携带write_disposition、schema_name、parent_table、resource_name等自定义属性subtype 为dlt Resourcesubtypes.py出口血缘OutletDataJob → 目标端 Dataset URN支持 Postgres、BigQuery、Snowflake 等平台依赖destination_platform_map完成缝合入口血缘Inlet用户手工配置的上游 Dataset URNdlt 自身不记录数据来源连接信息需通过source_dataset_urns/source_table_dataset_urns配置列级血缘直接拷贝型管线恰好 1 个 inlet 1 个 outlet以 FineGrainedLineage 形式输出字段级映射DataProcessInstance每次运行历史读_dlt_loads可选开启需 dlt 包与目标库凭据默认关闭1.1 dlt 如何存储元数据dlt 在每次pipeline.run()调用后会把管线状态写入本地目录典型结构如下~/.dlt/pipelines/ pipeline_name/ schemas/ schema_name.schema.yaml # 表定义列名、类型、write_disposition 等 state.json # 目标类型、数据集名、管线状态连接器就是围绕这份目录展开工作的dlt_client.py 中的DltClient.list_pipeline_names()遍历pipelines_dir下每个含schemas/子目录的子目录_read_schemas_from_filesystem()解析*.schema.yaml/*.schema.json文件文件回退路径state.json用于读取destination_type/destination_name与dataset_name这是构造出口 Dataset URN 的依据。注意schema 文件里的列类型会经 dlt_client.py 的DLT_TYPE_MAP映射为 DataHub 友好的类型名如text→string、bigint→long、bool→boolean、complex→map。dlt 注入的系统列_dlt_id、_dlt_load_id、_dlt_parent_id、_dlt_list_idx、_dlt_root_id会被识别并排除data_classes.py不会作为 DataHub schema 字段输出。二、前置条件让连接器能读到管线状态使用 dlt 连接器需要满足管线至少运行过一次dlt 会自动创建状态文件因此pipelines_dir中应已存在对应管线的schemas/与state.jsonpipelines_dir必须可从 DataHub 摄取进程访问连接器默认指向~/.dlt/pipelines可通过配置覆盖运行历史功能额外要求摄取环境中安装 dlt 包并配置目标库凭据见下文运行历史一节。2.1 如何确定你的pipelines_dirdlt 连接器将pipelines_dir默认值设为~/.dlt/pipelinesconfig.py配置解析时会自动展开~为用户主目录expand_user校验器。根据部署形态有三种常见取值本地 / Quickstart开箱即用pipelines_dir: ~/.dlt/pipelinesCI/CDGitHub Actions、Airflow、Jenkinsdlt 与 DataHub 摄取通常运行在不同 Job 中两者必须使用相同路径或共享存储pipelines_dir: /data/dlt-pipelines许多 dlt 用户已设置PIPELINES_DIR环境变量可结合 shell 默认值语法复用pipelines_dir: ${PIPELINES_DIR:-~/.dlt/pipelines}Kubernetes / Dockerdlt 与 DataHub 运行在不同 Pod 中需在两个 Pod 上挂载同一个 PersistentVolumeClaimpipelines_dir: /mnt/dlt-pipelines2.2 所需权限连接器只读本地文件基本元数据提取无需任何网络权限功能要求管线元数据DataFlow、DataJob、血缘对pipelines_dir具备文件系统读权限运行历史_dlt_loads安装 dlt 包 在~/.dlt/secrets.toml中配置目标库凭据三、最小可运行配置Recipe 全解析仓库提供了完整可运行的 recipe 示例 dlt_recipe.yml。以下是带注释的完整版可直接复制使用source: type: dlt config: # dlt 管线目录路径默认 ~/.dlt/pipelines。 # CI/CD 或容器环境下必须覆盖且需指向 dlt 实际写入的同一目录。 pipelines_dir: ~/.dlt/pipelines # 按管线名过滤匹配 dlt.pipeline() 传入的 pipeline_name。 pipeline_pattern: allow: - .* # deny: # - ^test_.* # 是否从 DataJob 向目标 Dataset URN 输出出口血缘。 include_lineage: true # 将 dlt 目标类型名映射为 DataHub 平台配置用于血缘缝合。 # env/platform_instance 必须与目标连接器使用的完全一致。 # 对使用三段式 URNdatabase.schema.table的 SQL 目标Postgres、Redshift 等 # database 字段为必填——dlt 只存储 schema 名。 destination_platform_map: postgres: database: my_database platform_instance: null env: PROD # bigquery: # platform_instance: my-gcp-project # env: PROD # snowflake: # platform_instance: my-account # env: PROD # 可选手工指定上游 Dataset URN。 # dlt 不记录数据源连接信息入口血缘必须在此配置。 # REST API 类源用 source_dataset_urns管线级作用于所有 DataJob # source_dataset_urns: # my_pipeline: # - urn:li:dataset:(urn:li:dataPlatform:salesforce,contacts,PROD) # # sql_database 类源用 source_table_dataset_urns表级 1:1 血缘 # source_table_dataset_urns: # my_pipeline: # my_table: # - urn:li:dataset:(urn:li:dataPlatform:postgres,prod_db.public.my_table,PROD) # 查询 _dlt_loads 并输出 DataProcessInstance 运行历史默认关闭。 # 需要 dlt 包 ~/.dlt/secrets.toml 或环境变量中的目标库凭据。 include_run_history: false # 运行历史时间窗口仅 include_run_history: true 时生效。 run_history_config: start_time: -7 days # end_time: now # 管线从 pipelines_dir 删除后自动移除对应 DataFlow/DataJob 实体。 stateful_ingestion: enabled: true remove_stale_metadata: true # 可选在 DataHub 中区分多个相互独立的 dlt 安装。 # platform_instance: my-dlt-instance env: PROD sink: type: datahub-rest config: server: http://localhost:8080四、配置参数逐项详解源码级所有配置项都由 pydantic 模型DltSourceConfig定义并校验config.py继承自StatefulIngestionConfigBase、PlatformInstanceConfigMixin、EnvConfigMixin因此天然支持env、platform_instance与状态化摄取等 DataHub 标准配置。参数默认值说明pipelines_dir~/.dlt/pipelinesdlt 管线状态目录配置加载时自动展开~pipeline_patternAllowDenyPattern.allow_all()按pipeline_name过滤管线的正则 allow/deny 模式include_lineagetrue是否输出 DataJob → 目标 Dataset URN 的出口血缘destination_platform_map{}目标类型 →DestinationPlatformConfig的映射用于构造血缘 URNsource_dataset_urns{}管线级入口 URN如 REST API 源key 为pipeline_namesource_table_dataset_urns{}表级入口 URN如 sql_database 源外层 key 为pipeline_name、内层为table_nameinclude_run_historyfalse是否查询目标端_dlt_loads并输出 DataProcessInstance 运行历史run_history_config空时间窗口运行历史时间窗口继承 DataHub 标准BaseTimeWindowConfig支持相对时间如-7 days与绝对 ISO 时间戳stateful_ingestionnull开启后管线从pipelines_dir删除时自动移除对应 DataFlow/DataJob4.1DestinationPlatformConfig每目标端 URN 构造配置destination_platform_map的每个值对应一个 DestinationPlatformConfig包含三个字段database可选字符串SQL 目标端的出口 Dataset URN 三段式前缀。dlt 只存储 schema 名dataset_name不存储数据库名因此 Postgres 等目标需手工提供platform_instance可选字符串目标平台在 DataHub 中的 platform instance必须与目标平台连接器使用的值完全一致env必填默认PROD目标环境PROD、DEV、STAGING 等。配置加载时会校验必须是 DataHub 的合法环境值——源码注释明确指出一个拼写错误的 env 会让出口 URN 与目标连接器 URN 无法缝合且表现为难以排查的血缘缺失因此在配置加载期快速失败比运行期报错更友好_env_must_be_valid校验器。源码层面_build_outlet_urnsdlt.py根据database是否配置决定表路径形态——配置了database时构造database.dataset_name.table_name三段式路径对应 Postgres/Redshift 等 SQL 目标否则构造dataset_name.table_name两段式对应 BigQuery/Snowflake 等云数仓项目/账户信息由platform_instance承载。随后经DatasetUrn.create_from_ids(platform_id, table_name, env, platform_instance)生成出口 URN。4.2 配置校验错误 URN 在启动期即被拒绝source_dataset_urns与source_table_dataset_urns都配置了字段校验器config.py会在datahub ingest启动时用DatasetUrn.from_string()逐个解析 URN任何格式非法的 URN 都会直接抛出ValidationError中止摄取而不是在运行时静默丢失入口血缘。五、数据建模为什么每个目标表对应一个 DataJob一个dlt.resource在返回嵌套数据时可能产生多个目标表——dlt 会按双下划线命名约定把 JSON 数组自动展开为子表例如orders与orders__items。连接器按目标表而非按 resource 输出 DataJob见 dlt.py 的表遍历逻辑这样每个目标表拥有干净的 1:1 出口血缘条目DataJob → 目标端 Dataset URN列级血缘保持在表粒度与下游血缘查询的预期一致在 DataHub 中浏览 dlt 管线时能看到所有已加载的表而非只有父 resource。其代价是这个 resource 产生了哪些表的抽象不再直接可见。为保留这一关联连接器通过自定义属性弥补每个子表 DataJob 携带parent_table自定义属性指向父表如orders__items上的parent_table: orders同一 resource 产出的所有表共享相同的resource_name自定义属性。需要特别说明的是父表与子表在 DataHub 血缘图中是兄弟关系同一批源数据行在加载时产生而非父子传递因此连接器不会在它们之间添加任何虚构的上游/下游血缘——那会歪曲真实数据流。集成测试的黄金文件 dlt_golden.json 中可以看到players_online_status、players_games、players_archives等 DataJob 均携带write_disposition、schema_name、resource_name自定义属性。六、血缘配置缝合到目标平台 Dataset6.1 出口血缘与缝合要求要让出口血缘连接到目标平台现有的 Dataset URNdestination_platform_map的env、platform_instance必须与目标平台连接器使用的环境与实例完全一致。如果 Postgres 连接器使用env: PROD且未配置platform_instancedestination_platform_map: postgres: env: PROD platform_instance: null database: my_database # 三段式 URN 必需database.schema.table为什么 SQL 目标端必须提供databasedlt 只存储 schema 名dataset_name而 DataHub 的 Postgres URN 采用三段式database.schema.table因此需要手工补充database才能与目标 URN 对齐。云数仓BigQuery、Snowflake则改用项目/账户作为platform_instancedestination_platform_map: bigquery: platform_instance: my-gcp-project env: PROD snowflake: platform_instance: my-account env: PROD6.2 入口血缘上游数据源dlt 不记录数据来自哪里——只记录数据去了哪里。因此上游血缘必须手工配置 Dataset URNREST API 类管线所有表共享同一上游source_dataset_urns: my_pipeline: - urn:li:dataset:(urn:li:dataPlatform:salesforce,contacts,PROD)sql_database 类管线每张表 1:1 对应源表source_table_dataset_urns: my_pipeline: my_table: - urn:li:dataset:(urn:li:dataPlatform:postgres,prod_db.public.my_table,PROD)源码中_build_inlet_urnsdlt.py会合并管线级与表级两类入口 URN——管线级 URN 作用于该管线下的所有 DataJob表级 URN 仅作用于指定表。6.3 列级血缘1:1 直接拷贝场景_build_fine_grained_lineagesdlt.py只在恰好 1 个 inlet 1 个 outlet时输出列级血缘——这是无歧义的 1:1 表拷贝场景典型如sql_database源。扇入/扇出场景因无法断言 1:1 列映射而不输出。dlt 注入的系统列_dlt_id、_dlt_load_id等会被排除因为它们在源端没有对应列。列级血缘以FineGrainedLineageClass形式逐列生成FIELD_SET→FIELD映射。七、运行历史DataProcessInstance 的可选能力运行历史需要在 DataHub 摄取环境中安装 dlt 包且目标库凭据可访问pip install dlt[postgres] # 或 dlt[bigquery]、dlt[snowflake] 等凭据从 dlt 的标准位置~/.dlt/secrets.toml或环境变量读取export DESTINATION__POSTGRES__CREDENTIALS__HOSTlocalhost export DESTINATION__POSTGRES__CREDENTIALS__DATABASEmy_db export DESTINATION__POSTGRES__CREDENTIALS__USERNAMEdlt export DESTINATION__POSTGRES__CREDENTIALS__PASSWORDsecretrun_history_config时间窗口会被严格遵守——用start_time/end_time限制摄取哪些 loadinclude_run_history: true run_history_config: start_time: -7 days end_time: now底层实现dlt_client.pyget_run_history()通过pipeline.sql_client()在目标端执行SELECT load_id, schema_name, status, inserted_at, schema_version_hash FROM _dlt_loads ORDER BY inserted_at DESC逐行解析后在 dlt.py 中为每次 load 生成一个 DataProcessInstance含 start/end 事件。_dlt_loads中status0LOADED映射为成功运行其他状态码一律映射为失败以保留审计信号_DLT_LOAD_STATUS_MAPdlt.py。当 dlt 未安装时连接器会在报告中给出Run history unavailable警告并跳过该功能但 DataFlow / DataJob / 出口血缘照常输出——运行历史失败不会拖垮整个摄取。八、能力矩阵与限制8.1 能力概览能力状态说明DataFlow / DataJob始终支持每个管线一个 DataFlow每个目标表一个 DataJob出口血缘始终支持include_lineage: true时需要destination_platform_map与目标连接器匹配入口血缘用户配置dlt 不存储源身份通过source_dataset_urns配置列级血缘部分支持仅限恰好 1 inlet 1 outlet 的表无歧义 1:1 拷贝运行历史DataProcessInstance可选需include_run_history: true dlt 包 目标库凭据删除检测状态化摄取支持管线从pipelines_dir删除后移除对应 DataFlow/DataJob属主Ownership不支持dlt 状态中不包含属主信息8.2 已知限制入口血缘必须手工配置dlt 状态文件不存储源系统连接信息上游 Dataset URN 只能通过source_dataset_urns或source_table_dataset_urns手工指定运行历史依赖目标库访问查询_dlt_loads需要 dlt 包与目标凭据。dlt 未安装或凭据缺失时连接器仍输出 DataFlow / DataJob / 出口血缘但跳过运行历史不支持属主信息dlt 不记录管线属主。九、故障排查没有输出任何实体检查pipelines_dir是否指向包含带schemas/子目录的子目录运行datahub ingest -c recipe.yml --test-source-connection验证路径可读——连接器的test_connectiondlt.py会校验目录存在性并检测 dlt 包是否安装未安装时提示使用文件系统回退血缘能力不受影响。血缘无法缝合核对destination_platform_map的 env/instance/database 是否与目标连接器使用的一致在 DataHub 中检查目标 Dataset URN与 dlt 连接器构造的 URN 对比Postgres 目标端务必确认destination_platform_map.postgres中设置了database。运行历史为空确认include_run_history: true已设置确认 dlt 包已安装python -c import dlt; print(dlt.__version__)确认目标凭据在~/.dlt/secrets.toml或环境变量中检查 DataHub 摄取日志中 dlt 连接器的警告run_history_errors等计数可在摄取报告中查看见 dlt_report.py。嵌套子表如orders__itemsdlt 会把嵌套 JSON 按双下划线命名自动展开为子表它们在 DataHub 中表现为独立的 DataJob自定义属性中带有parent_table。这是预期行为无需处理。十、从源码与测试验证行为主流程DltSource.get_workunits_internaldlt.py发现管线 →pipeline_pattern过滤 →get_pipeline_info读取元数据 → 逐 schema 逐表产出 DataJobdlt 系统表_dlt_loads、_dlt_version、_dlt_pipeline_state会被跳过DLT_SYSTEM_TABLESdlt.py双路径读取优先 dlt SDK attachdlt.pipeline(...)仅恢复状态、不触发运行失败自动回退文件系统解析dlt_client.py测试佐证集成测试 test_dlt.py 使用预置的 schema YAML 夹具模拟真实 dlt 状态输出无需安装 dlt 即可跑通黄金文件 dlt_golden.json 记录了 DataFlow/DataJob 的完整期望输出。总而言之DataHub dlt 连接器以零网络依赖读本地状态为设计核心用 DataFlow DataJob 三类血缘把 dlt 管线完整映射进 DataHub 元数据体系只要把pipelines_dir指对、把destination_platform_map配准即可在 DataHub 中浏览、检索并追溯每条 dlt 管线的数据流动全貌。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
DataHub dlt 连接器实战从 dlt 本地状态目录提取 Pipeline、DataJob 与血缘元数据【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub导读本文讲解 DataHub 内置的dltdata load tool摄取连接器它直接读取 dlt 写入本地的管线状态目录默认~/.dlt/pipelines/无需实时连接 dlt 或目标库即可提取 Pipeline 元数据并将 dlt 管线建模为 DataFlow、每个目标表建模为 DataJob同时输出出口Outlet与入口Inlet血缘、列级血缘以及可选的运行历史。读完本文你将掌握 dlt 连接器的数据模型、pipelines_dir在不同部署形态下的配置方式、血缘缝合的关键参数以及运行历史、状态化摄取等进阶能力的启用方法。一、连接器概览dlt 模块摄取什么dltdata load tool是面向数据加载的 Python 库DataHub 的 dlt 连接器从 dlt 的本地状态目录读取管线元数据并建模为 DataHub 实体dlt.py 是连接器主入口类DltSource标注平台名dlt、支持状态BETA。基本元数据提取不需要与 dlt 或目标数据源建立实时连接——连接器只读取本地文件。若摄取环境中安装了 dlt Python 包则优先走 dlt SDK 获取更丰富的元数据否则回退为直接解析 schema YAML 状态文件。连接器摄取的核心实体与血缘包括实体建模粒度说明DataFlow每个 dlt 管线一个pipeline_name通过_build_dataflow生成携带destination、dataset_name、pipelines_dir、working_dir等自定义属性subtype 为dlt Pipeline见 subtypes.pyDataJob每个目标表一个含自动展开的子表如orders__items通过_build_datajob生成携带write_disposition、schema_name、parent_table、resource_name等自定义属性subtype 为dlt Resourcesubtypes.py出口血缘OutletDataJob → 目标端 Dataset URN支持 Postgres、BigQuery、Snowflake 等平台依赖destination_platform_map完成缝合入口血缘Inlet用户手工配置的上游 Dataset URNdlt 自身不记录数据来源连接信息需通过source_dataset_urns/source_table_dataset_urns配置列级血缘直接拷贝型管线恰好 1 个 inlet 1 个 outlet以 FineGrainedLineage 形式输出字段级映射DataProcessInstance每次运行历史读_dlt_loads可选开启需 dlt 包与目标库凭据默认关闭1.1 dlt 如何存储元数据dlt 在每次pipeline.run()调用后会把管线状态写入本地目录典型结构如下~/.dlt/pipelines/ pipeline_name/ schemas/ schema_name.schema.yaml # 表定义列名、类型、write_disposition 等 state.json # 目标类型、数据集名、管线状态连接器就是围绕这份目录展开工作的dlt_client.py 中的DltClient.list_pipeline_names()遍历pipelines_dir下每个含schemas/子目录的子目录_read_schemas_from_filesystem()解析*.schema.yaml/*.schema.json文件文件回退路径state.json用于读取destination_type/destination_name与dataset_name这是构造出口 Dataset URN 的依据。注意schema 文件里的列类型会经 dlt_client.py 的DLT_TYPE_MAP映射为 DataHub 友好的类型名如text→string、bigint→long、bool→boolean、complex→map。dlt 注入的系统列_dlt_id、_dlt_load_id、_dlt_parent_id、_dlt_list_idx、_dlt_root_id会被识别并排除data_classes.py不会作为 DataHub schema 字段输出。二、前置条件让连接器能读到管线状态使用 dlt 连接器需要满足管线至少运行过一次dlt 会自动创建状态文件因此pipelines_dir中应已存在对应管线的schemas/与state.jsonpipelines_dir必须可从 DataHub 摄取进程访问连接器默认指向~/.dlt/pipelines可通过配置覆盖运行历史功能额外要求摄取环境中安装 dlt 包并配置目标库凭据见下文运行历史一节。2.1 如何确定你的pipelines_dirdlt 连接器将pipelines_dir默认值设为~/.dlt/pipelinesconfig.py配置解析时会自动展开~为用户主目录expand_user校验器。根据部署形态有三种常见取值本地 / Quickstart开箱即用pipelines_dir: ~/.dlt/pipelinesCI/CDGitHub Actions、Airflow、Jenkinsdlt 与 DataHub 摄取通常运行在不同 Job 中两者必须使用相同路径或共享存储pipelines_dir: /data/dlt-pipelines许多 dlt 用户已设置PIPELINES_DIR环境变量可结合 shell 默认值语法复用pipelines_dir: ${PIPELINES_DIR:-~/.dlt/pipelines}Kubernetes / Dockerdlt 与 DataHub 运行在不同 Pod 中需在两个 Pod 上挂载同一个 PersistentVolumeClaimpipelines_dir: /mnt/dlt-pipelines2.2 所需权限连接器只读本地文件基本元数据提取无需任何网络权限功能要求管线元数据DataFlow、DataJob、血缘对pipelines_dir具备文件系统读权限运行历史_dlt_loads安装 dlt 包 在~/.dlt/secrets.toml中配置目标库凭据三、最小可运行配置Recipe 全解析仓库提供了完整可运行的 recipe 示例 dlt_recipe.yml。以下是带注释的完整版可直接复制使用source: type: dlt config: # dlt 管线目录路径默认 ~/.dlt/pipelines。 # CI/CD 或容器环境下必须覆盖且需指向 dlt 实际写入的同一目录。 pipelines_dir: ~/.dlt/pipelines # 按管线名过滤匹配 dlt.pipeline() 传入的 pipeline_name。 pipeline_pattern: allow: - .* # deny: # - ^test_.* # 是否从 DataJob 向目标 Dataset URN 输出出口血缘。 include_lineage: true # 将 dlt 目标类型名映射为 DataHub 平台配置用于血缘缝合。 # env/platform_instance 必须与目标连接器使用的完全一致。 # 对使用三段式 URNdatabase.schema.table的 SQL 目标Postgres、Redshift 等 # database 字段为必填——dlt 只存储 schema 名。 destination_platform_map: postgres: database: my_database platform_instance: null env: PROD # bigquery: # platform_instance: my-gcp-project # env: PROD # snowflake: # platform_instance: my-account # env: PROD # 可选手工指定上游 Dataset URN。 # dlt 不记录数据源连接信息入口血缘必须在此配置。 # REST API 类源用 source_dataset_urns管线级作用于所有 DataJob # source_dataset_urns: # my_pipeline: # - urn:li:dataset:(urn:li:dataPlatform:salesforce,contacts,PROD) # # sql_database 类源用 source_table_dataset_urns表级 1:1 血缘 # source_table_dataset_urns: # my_pipeline: # my_table: # - urn:li:dataset:(urn:li:dataPlatform:postgres,prod_db.public.my_table,PROD) # 查询 _dlt_loads 并输出 DataProcessInstance 运行历史默认关闭。 # 需要 dlt 包 ~/.dlt/secrets.toml 或环境变量中的目标库凭据。 include_run_history: false # 运行历史时间窗口仅 include_run_history: true 时生效。 run_history_config: start_time: -7 days # end_time: now # 管线从 pipelines_dir 删除后自动移除对应 DataFlow/DataJob 实体。 stateful_ingestion: enabled: true remove_stale_metadata: true # 可选在 DataHub 中区分多个相互独立的 dlt 安装。 # platform_instance: my-dlt-instance env: PROD sink: type: datahub-rest config: server: http://localhost:8080四、配置参数逐项详解源码级所有配置项都由 pydantic 模型DltSourceConfig定义并校验config.py继承自StatefulIngestionConfigBase、PlatformInstanceConfigMixin、EnvConfigMixin因此天然支持env、platform_instance与状态化摄取等 DataHub 标准配置。参数默认值说明pipelines_dir~/.dlt/pipelinesdlt 管线状态目录配置加载时自动展开~pipeline_patternAllowDenyPattern.allow_all()按pipeline_name过滤管线的正则 allow/deny 模式include_lineagetrue是否输出 DataJob → 目标 Dataset URN 的出口血缘destination_platform_map{}目标类型 →DestinationPlatformConfig的映射用于构造血缘 URNsource_dataset_urns{}管线级入口 URN如 REST API 源key 为pipeline_namesource_table_dataset_urns{}表级入口 URN如 sql_database 源外层 key 为pipeline_name、内层为table_nameinclude_run_historyfalse是否查询目标端_dlt_loads并输出 DataProcessInstance 运行历史run_history_config空时间窗口运行历史时间窗口继承 DataHub 标准BaseTimeWindowConfig支持相对时间如-7 days与绝对 ISO 时间戳stateful_ingestionnull开启后管线从pipelines_dir删除时自动移除对应 DataFlow/DataJob4.1DestinationPlatformConfig每目标端 URN 构造配置destination_platform_map的每个值对应一个 DestinationPlatformConfig包含三个字段database可选字符串SQL 目标端的出口 Dataset URN 三段式前缀。dlt 只存储 schema 名dataset_name不存储数据库名因此 Postgres 等目标需手工提供platform_instance可选字符串目标平台在 DataHub 中的 platform instance必须与目标平台连接器使用的值完全一致env必填默认PROD目标环境PROD、DEV、STAGING 等。配置加载时会校验必须是 DataHub 的合法环境值——源码注释明确指出一个拼写错误的 env 会让出口 URN 与目标连接器 URN 无法缝合且表现为难以排查的血缘缺失因此在配置加载期快速失败比运行期报错更友好_env_must_be_valid校验器。源码层面_build_outlet_urnsdlt.py根据database是否配置决定表路径形态——配置了database时构造database.dataset_name.table_name三段式路径对应 Postgres/Redshift 等 SQL 目标否则构造dataset_name.table_name两段式对应 BigQuery/Snowflake 等云数仓项目/账户信息由platform_instance承载。随后经DatasetUrn.create_from_ids(platform_id, table_name, env, platform_instance)生成出口 URN。4.2 配置校验错误 URN 在启动期即被拒绝source_dataset_urns与source_table_dataset_urns都配置了字段校验器config.py会在datahub ingest启动时用DatasetUrn.from_string()逐个解析 URN任何格式非法的 URN 都会直接抛出ValidationError中止摄取而不是在运行时静默丢失入口血缘。五、数据建模为什么每个目标表对应一个 DataJob一个dlt.resource在返回嵌套数据时可能产生多个目标表——dlt 会按双下划线命名约定把 JSON 数组自动展开为子表例如orders与orders__items。连接器按目标表而非按 resource 输出 DataJob见 dlt.py 的表遍历逻辑这样每个目标表拥有干净的 1:1 出口血缘条目DataJob → 目标端 Dataset URN列级血缘保持在表粒度与下游血缘查询的预期一致在 DataHub 中浏览 dlt 管线时能看到所有已加载的表而非只有父 resource。其代价是这个 resource 产生了哪些表的抽象不再直接可见。为保留这一关联连接器通过自定义属性弥补每个子表 DataJob 携带parent_table自定义属性指向父表如orders__items上的parent_table: orders同一 resource 产出的所有表共享相同的resource_name自定义属性。需要特别说明的是父表与子表在 DataHub 血缘图中是兄弟关系同一批源数据行在加载时产生而非父子传递因此连接器不会在它们之间添加任何虚构的上游/下游血缘——那会歪曲真实数据流。集成测试的黄金文件 dlt_golden.json 中可以看到players_online_status、players_games、players_archives等 DataJob 均携带write_disposition、schema_name、resource_name自定义属性。六、血缘配置缝合到目标平台 Dataset6.1 出口血缘与缝合要求要让出口血缘连接到目标平台现有的 Dataset URNdestination_platform_map的env、platform_instance必须与目标平台连接器使用的环境与实例完全一致。如果 Postgres 连接器使用env: PROD且未配置platform_instancedestination_platform_map: postgres: env: PROD platform_instance: null database: my_database # 三段式 URN 必需database.schema.table为什么 SQL 目标端必须提供databasedlt 只存储 schema 名dataset_name而 DataHub 的 Postgres URN 采用三段式database.schema.table因此需要手工补充database才能与目标 URN 对齐。云数仓BigQuery、Snowflake则改用项目/账户作为platform_instancedestination_platform_map: bigquery: platform_instance: my-gcp-project env: PROD snowflake: platform_instance: my-account env: PROD6.2 入口血缘上游数据源dlt 不记录数据来自哪里——只记录数据去了哪里。因此上游血缘必须手工配置 Dataset URNREST API 类管线所有表共享同一上游source_dataset_urns: my_pipeline: - urn:li:dataset:(urn:li:dataPlatform:salesforce,contacts,PROD)sql_database 类管线每张表 1:1 对应源表source_table_dataset_urns: my_pipeline: my_table: - urn:li:dataset:(urn:li:dataPlatform:postgres,prod_db.public.my_table,PROD)源码中_build_inlet_urnsdlt.py会合并管线级与表级两类入口 URN——管线级 URN 作用于该管线下的所有 DataJob表级 URN 仅作用于指定表。6.3 列级血缘1:1 直接拷贝场景_build_fine_grained_lineagesdlt.py只在恰好 1 个 inlet 1 个 outlet时输出列级血缘——这是无歧义的 1:1 表拷贝场景典型如sql_database源。扇入/扇出场景因无法断言 1:1 列映射而不输出。dlt 注入的系统列_dlt_id、_dlt_load_id等会被排除因为它们在源端没有对应列。列级血缘以FineGrainedLineageClass形式逐列生成FIELD_SET→FIELD映射。七、运行历史DataProcessInstance 的可选能力运行历史需要在 DataHub 摄取环境中安装 dlt 包且目标库凭据可访问pip install dlt[postgres] # 或 dlt[bigquery]、dlt[snowflake] 等凭据从 dlt 的标准位置~/.dlt/secrets.toml或环境变量读取export DESTINATION__POSTGRES__CREDENTIALS__HOSTlocalhost export DESTINATION__POSTGRES__CREDENTIALS__DATABASEmy_db export DESTINATION__POSTGRES__CREDENTIALS__USERNAMEdlt export DESTINATION__POSTGRES__CREDENTIALS__PASSWORDsecretrun_history_config时间窗口会被严格遵守——用start_time/end_time限制摄取哪些 loadinclude_run_history: true run_history_config: start_time: -7 days end_time: now底层实现dlt_client.pyget_run_history()通过pipeline.sql_client()在目标端执行SELECT load_id, schema_name, status, inserted_at, schema_version_hash FROM _dlt_loads ORDER BY inserted_at DESC逐行解析后在 dlt.py 中为每次 load 生成一个 DataProcessInstance含 start/end 事件。_dlt_loads中status0LOADED映射为成功运行其他状态码一律映射为失败以保留审计信号_DLT_LOAD_STATUS_MAPdlt.py。当 dlt 未安装时连接器会在报告中给出Run history unavailable警告并跳过该功能但 DataFlow / DataJob / 出口血缘照常输出——运行历史失败不会拖垮整个摄取。八、能力矩阵与限制8.1 能力概览能力状态说明DataFlow / DataJob始终支持每个管线一个 DataFlow每个目标表一个 DataJob出口血缘始终支持include_lineage: true时需要destination_platform_map与目标连接器匹配入口血缘用户配置dlt 不存储源身份通过source_dataset_urns配置列级血缘部分支持仅限恰好 1 inlet 1 outlet 的表无歧义 1:1 拷贝运行历史DataProcessInstance可选需include_run_history: true dlt 包 目标库凭据删除检测状态化摄取支持管线从pipelines_dir删除后移除对应 DataFlow/DataJob属主Ownership不支持dlt 状态中不包含属主信息8.2 已知限制入口血缘必须手工配置dlt 状态文件不存储源系统连接信息上游 Dataset URN 只能通过source_dataset_urns或source_table_dataset_urns手工指定运行历史依赖目标库访问查询_dlt_loads需要 dlt 包与目标凭据。dlt 未安装或凭据缺失时连接器仍输出 DataFlow / DataJob / 出口血缘但跳过运行历史不支持属主信息dlt 不记录管线属主。九、故障排查没有输出任何实体检查pipelines_dir是否指向包含带schemas/子目录的子目录运行datahub ingest -c recipe.yml --test-source-connection验证路径可读——连接器的test_connectiondlt.py会校验目录存在性并检测 dlt 包是否安装未安装时提示使用文件系统回退血缘能力不受影响。血缘无法缝合核对destination_platform_map的 env/instance/database 是否与目标连接器使用的一致在 DataHub 中检查目标 Dataset URN与 dlt 连接器构造的 URN 对比Postgres 目标端务必确认destination_platform_map.postgres中设置了database。运行历史为空确认include_run_history: true已设置确认 dlt 包已安装python -c import dlt; print(dlt.__version__)确认目标凭据在~/.dlt/secrets.toml或环境变量中检查 DataHub 摄取日志中 dlt 连接器的警告run_history_errors等计数可在摄取报告中查看见 dlt_report.py。嵌套子表如orders__itemsdlt 会把嵌套 JSON 按双下划线命名自动展开为子表它们在 DataHub 中表现为独立的 DataJob自定义属性中带有parent_table。这是预期行为无需处理。十、从源码与测试验证行为主流程DltSource.get_workunits_internaldlt.py发现管线 →pipeline_pattern过滤 →get_pipeline_info读取元数据 → 逐 schema 逐表产出 DataJobdlt 系统表_dlt_loads、_dlt_version、_dlt_pipeline_state会被跳过DLT_SYSTEM_TABLESdlt.py双路径读取优先 dlt SDK attachdlt.pipeline(...)仅恢复状态、不触发运行失败自动回退文件系统解析dlt_client.py测试佐证集成测试 test_dlt.py 使用预置的 schema YAML 夹具模拟真实 dlt 状态输出无需安装 dlt 即可跑通黄金文件 dlt_golden.json 记录了 DataFlow/DataJob 的完整期望输出。总而言之DataHub dlt 连接器以零网络依赖读本地状态为设计核心用 DataFlow DataJob 三类血缘把 dlt 管线完整映射进 DataHub 元数据体系只要把pipelines_dir指对、把destination_platform_map配准即可在 DataHub 中浏览、检索并追溯每条 dlt 管线的数据流动全貌。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
