拓冰网站定制 
用 dlt 构建 Chess API 生产级数据流水线重试、通知、可观测性与自动化测试实战【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt本技术指南基于开源数据加载工具 dltdata load tool官方示例 chess_production.md带你从零构建一条对接 Chess.com 公共 API 的生产级数据流水线。示例脚本展示了如何提取棋手资料与对局数据并把「重试机制、Slack 通知、加载结果检查、模式演进告警、自动化测试、LoadInfo/Trace 回写目标库」等生产环境必备能力全部落地到同一段可运行的代码中。读完本文你将掌握 dlt 中source/transformer组合建模、基于 tenacity 的分阶段重试、sql_client与normalize_info两种数据校验方式以及如何把流水线运行元数据作为数据本身存回目标数据仓库。示例概览这条流水线做了什么该示例是一个完整的 Python 脚本与 Chess.com 公共 APIhttps://api.chess.com/pub/交互抽取棋手列表按头衔如 GM及其个人资料、月度对局数据最终写入 duckdb 数据库。它的特殊之处在于不是为了演示“能跑通”而是演示“上线之后怎么管理”——脚本在生产运行流程中依次完成以下工作加载完成后检查加载包load package逐个查看其中的 job 与 schema 变更把加载信息load info、schema 更新、运行 trace作为数据回写目标库形成可查询的元数据表在**模式演进schema evolution**发生时触发通知使用上下文管理器与 tenacity 重试策略独立重试流水线中的不同阶段借助sql_client与normalize_info编写基础数据校验测试如检查表行数并断言。官方文档给出的学习目标在 chess_production.md 中列出本文按脚本的实际执行顺序逐段拆解。数据源建模source resource transformer 的组合示例的核心数据源定义在dlt.source装饰的chess()函数中chess_production.md它把三个资源组合成一条完整的数据 DAGdlt.source def chess( chess_url: str dlt.config.value, title: str GM, max_players: int 2, year: int 2022, month: int 10, ) - Any: def _get_data_with_retry(path: str) - StrAny: r client.get(f{chess_url}{path}) return r.json()通过dlt.config.value注入外部配置注意第一个参数chess_url: str dlt.config.valuedlt 会从环境变量、TOML 配置等来源注入该值不需要在代码里写死 API 地址。这就是 dlt 的配置注入机制与之配套的title、max_players、year、month都是普通默认参数可在调用时覆盖。并行化 transformerplayers_profilesdlt.transformer(data_fromplayers, write_dispositionreplace, parallelizedTrue) def players_profiles(username: Any) - TDataItems: print(fgetting {username} profile via thread {threading.current_thread().name}) sleep(1) # add some latency to show parallel runs return _get_data_with_retry(fplayer/{username})players资源逐个 yield 棋手用户名yield from ...[players][:max_players]players_profiles作为 transformer 消费这些用户名并抓取各自的 profile。parallelizedTrue让 transformer 在线程池中并行运行代码中的threading.current_thread().name与sleep(1)正是为了在运行时直观展示并行效果这是 dlt 提供的资源级并发能力。容忍缺失数据的 transformerplayers_gamesdlt.transformer(data_fromplayers, write_dispositionappend) def players_games(username: Any) - Iterator[TDataItems]: path fplayer/{username}/games/{year:04d}/{month:02d} try: yield _get_data_with_retry(path)[games] except requests.HTTPError as exc: # we allow players to not have games for some months if not exc.response.status_code 404: raise excplayers_games抓取指定年月的对局但对 404 做了显式容忍——某些棋手在某月确实没有对局记录这不是错误。这一模式在生产中非常常见对上游 API 的业务性缺失与真正的故障要区别对待。三个资源使用了两种写入策略players/players_profiles用replace全量替换players_games用append增量追加配合使用能让数据表各取所需。关于并行化 transformer 与资源编排的更多通用用法可参考 transformers.mddlt 的 source 装饰器与资源系统定义位于 dlt/extract/source.py 与 dlt/extract/resource.py。生产级重试tenacity 与 retry_load 组合拳数据流水线上线后网络抖动、目标库临时不可用都是常态。示例没有把重试逻辑写死在pipeline.run外面套一个简单循环而是使用 tenacity 的Retrying对象包住每次运行尝试并配合 dlt 提供的retry_load判定函数chess_production.mddef load_data_with_retry(pipeline, data): try: for attempt in Retrying( stopstop_after_attempt(5), waitwait_exponential(multiplier1.5, min4, max10), retryretry_if_exception(retry_load(())), reraiseTrue, ): with attempt: logger.info( fRunning the pipeline, attempt{attempt.retry_state.attempt_number} ) load_info pipeline.run(data) logger.info(str(load_info)) send_slack_message( pipeline.runtime_config.slack_incoming_hook, Data was successfully loaded!, ) except Exception: # we get here after all the failed retries send_slack_message( pipeline.runtime_config.slack_incoming_hook, Something went wrong! ) raise这里load_data_with_retry是自定义辅助函数不是 dlt 内置 API它的重试策略参数有明确语义参数值含义stopstop_after_attempt(5)最多尝试 5 次waitwait_exponential(multiplier1.5, min4, max10)指数退避等待基础乘数 1.5等待 4~10 秒retryretry_if_exception(retry_load(()))只有满足retry_load判定条件的异常才重试reraiseTrue重试耗尽后抛出原始异常retry_load 的判定逻辑什么该重试什么不该重试retry_load是 dlt 提供的内置重试策略工厂实现在 dlt/pipeline/helpers.py。其核心逻辑是只重试load阶段如果异常是PipelineStepFailed且出错的步骤不在retry_on_pipeline_steps默认只有load中则不重试——extract、normalize 阶段失败通常不是网络抖动重试没有意义不重试终态异常TerminalException配置缺失、认证失败、job 终态失败等异常重复多少次都不会成功见 dlt/pipeline/helpers.py 对TerminalException的判断。该函数的 docstring 给出了等价的最小用法示例dlt/pipeline/helpers.pyfor attempt in Retrying(stopstop_after_attempt(3), retryretry_if_exception(retry_load(())), reraiseTrue): with attempt: p.run(data)retry_load(())中的空元组表示只重试load步骤本身等价于默认(load,)。如果你希望其他阶段也参与重试可以传入如retry_load((load, extract))如果需要处理“多流水线并行时 schema 竞争创建表/加列失败”的场景dlt 还提供了retry_schema_update()策略可与retry_load用|组合见 dlt/pipeline/helpers.py。加载结果的深度检查LoadInfo 与 load_packages成功重试并完成一次pipeline.run后返回的load_info是LoadInfo类型——dlt 定义的“最近一次加载包信息”元组由pipeline.run和load方法返回dlt/common/pipeline.py。示例用它做了四件事chess_production.md# see when load was started logger.info(fPipeline was started: {load_info.started_at}) # print the information on the first load package and all jobs inside logger.info(fFirst load package info: {load_info.load_packages[0]}) # print the information on the first completed job in first load package logger.info( fFirst completed job info: {load_info.load_packages[0].jobs[completed_jobs][0]} ) # check for schema updates: schema_updates [p.schema_update for p in load_info.load_packages] if schema_updates: send_slack_message( pipeline.runtime_config.slack_incoming_hook, Schema was updated! )要点拆解load_info.started_at/finished_at本次 load 阶段的起止时间LoadInfo.asstr()会自动格式化输出“load 阶段耗时、加载了多少个包”等人类可读摘要dlt/common/pipeline.pyload_info.load_packagesLoadPackageInfo列表定义于 dlt/common/storages/load_package.py每个包对应一次独立的加载批次包内有完整的 job 状态分类jobs[completed_jobs]等package.schema_update本次加载引起的 schema 变更集合。只要非空就说明上游数据让 dlt 自动演进出了新表或新列——这是数据漂移的早期信号示例选择立即向 Slack 发送 “Schema was updated!” 通知。这样就把“数据有没有变化、变化了什么”变成可观测、可告警的信号。用 sql_client 与 normalize_info 做自动化校验生产流水线需要在加载后做基本的数据质量断言。示例提供了两种互补的校验手段chess_production.md方式一sql_client 直接查目标库with pipeline.sql_client() as client: with client.execute_query(SELECT COUNT(*) FROM players) as cursor: count cursor.fetchone()[0] if count 0: logger.info(Warning: No data in players table) else: logger.info(fPlayers table contains {count} rows) assert count MAX_PLAYERSpipeline.sql_client()是 dlt 在 dlt/pipeline/pipeline.py 中提供的标准入口它返回一个已认证且默认限定在流水线 dataset上的 SQL 客户端用with语句管理连接生命周期。其 docstring 给出了通用查询示例含参数化查询with pipeline.sql_client() as client: with client.execute_query( SELECT id, name, email FROM customers WHERE id %s, 10 ) as cursor: print(cursor.fetchall())需要注意的是sql_client只在目标库支持 SQL 接口时可用例如 duckdb、postgres、snowflake 等对于纯文件类目标如 filesystem会抛出SqlClientNotAvailable见 dlt/pipeline/pipeline.py。方式二normalize_info 直接读元数据normalize_info pipeline.last_trace.last_normalize_info count normalize_info.row_counts.get(players, 0) if count 0: logger.info(Warning: No data in players table) else: logger.info(fPlayers table contains {count} rows) assert count MAX_PLAYERSpipeline.last_trace属性从标准位置加载最近一次运行 tracedlt/pipeline/pipeline.pylast_normalize_info从 trace 中取出 normalize 阶段的NormalizeInfodlt/pipeline/trace.py。NormalizeInfo.row_counts属性按表汇总了本次归一化处理的数据项数量dlt/common/pipeline.py。两种方式各有优势sql_client校验的是目标库里的真实数据端到端最可靠normalize_info校验的是归一化阶段的处理结果不需要真正查询目标库成本低、与目标无关。示例把两者都断言为MAX_PLAYERS脚本顶部定义为 5见 chess_production.md这样任何一步丢失数据都会立刻让测试失败。把运行元数据回写目标库load_info、trace 与 schema 更新表示例最具“生产可观测性”特色的一步是把流水线自身的运行元数据当作普通数据加载回同一个目标库chess_production.md# we reuse the pipeline instance below and load to the same dataset as data logger.info(Saving the load info in the destination) pipeline.run([load_info], table_name_load_info) assert _load_info in pipeline.last_trace.last_normalize_info.row_counts # save trace to destination, sensitive data will be removed logger.info(Saving the trace in the destination) pipeline.run([pipeline.last_trace], table_name_trace) assert _trace in pipeline.last_trace.last_normalize_info.row_counts这里有三个关键点复用 pipeline 实例再次调用pipeline.run时数据会加载到与业务数据相同的 dataset 中只是换成了_load_info、_trace这样的下划线前缀表名与业务表区分开LoadInfo/Trace本身就是可加载对象dlt 为它们实现了asdict()序列化LoadInfo.asdict见 dlt/common/pipeline.py会展开为带load_id的job_metrics、outputs明细行所以可以直接塞进pipeline.runtrace 在落库时会自动移除敏感数据文档注释明确说明 “sensitive data will be removed”用 row_counts 断言落库成功_load_info in pipeline.last_trace.last_normalize_info.row_counts确认元数据表确实写入了。接着示例把本次 schema 演进的明细也存了下来chess_production.md# print all the new tables/columns in for package in load_info.load_packages: for table_name, table in package.schema_update.items(): logger.info(fTable {table_name}: {table.get(description)}) for column_name, column in table[columns].items(): logger.info(f\tcolumn {column_name}: {column[data_type]}) # save the new tables and column schemas to the destination: table_updates [p.asdict()[tables] for p in load_info.load_packages] pipeline.run(table_updates, table_name_new_tables) assert _new_tables in pipeline.last_trace.last_normalize_info.row_counts先在日志中打印新增的表名、描述、列名与数据类型再把load_packages的asdict()[tables]结构整体写入_new_tables表。这样模式变更本身就成了可查询的数据——将来排查“哪次运行引入了哪张新表/新列”直接查_new_tables即可配合_load_info、_trace就能还原每次运行的全貌。运行方式与完整脚本脚本入口在 chess_production.mdif __name__ __main__: # create dlt pipeline pipeline dlt.pipeline( pipeline_namechess_pipeline, destinationduckdb, dataset_namechess_data, ) # get data for a few famous players data chess(max_playersMAX_PLAYERS) load_data_with_retry(pipeline, data)运行前提安装 dlt 及 duckdb 目标支持pip install dlt[duckdb]另需requests、tenacity依赖通过dlt.pipeline(pipeline_namechess_pipeline, destinationduckdb, dataset_namechess_data)创建流水线——pipeline_name用于本地状态/trace 存储destination指定目标库类型dataset_name指定目标数据集chess(max_playersMAX_PLAYERS)实例化数据源load_data_with_retry负责带重试地运行并执行全部检查与回写chess_url未在代码中赋值需通过 dlt 配置注入例如设置环境变量或用 TOML 配置例如CHESS__CHESS_URL环境变量对应chess_url参数dlt 按资源名__参数名的层级自动解析。运行后预期观察到的现象与源码逻辑一一对应日志中出现多次 “Running the pipeline, attempt1/2/…”仅在遇到可重试异常时出现多次players_profiles的打印中线程名各不相同并行化生效日志输出LoadInfo摘要、started_at、首个 load package 与其 completed jobs 明细若发生 schema 演进Slack 收到 “Schema was updated!”目标库chess_data数据集中新增players、players_profiles、players_games业务表以及_load_info、_trace、_new_tables三张元数据表assert count MAX_PLAYERS保证 players 表行数与预期严格一致。如果想观察“并行化 逐项处理”的 transformer 行为可以把max_players调大_get_data_with_retry复用了 dlt 内置的带重试 HTTP 客户端dlt.sources.helpers.requests.client见 dlt/sources/helpers/requests它对公共 API 的瞬时故障有额外一层兜底。小结一条生产流水线的完整检查清单把 chess_production.md 中演示的能力整理成清单可直接作为你自己 dlt 流水线的上线模板建模用dlt.sourcedlt.resourcedlt.transformer组合数据 DAG对可并行任务开parallelizedTrue用write_disposition区分replace/append语义对上游 404 等业务性缺失显式容忍配置敏感地址与密钥通过dlt.config.value从环境变量/TOML 注入不写死在代码里重试用 tenacityRetrying包裹运行配合retry_load只重试load阶段、避开终态异常必要时用retry_schema_update应对并发 schema 竞争告警成功、失败、schema 演进三类事件通过send_slack_message通知到pipeline.runtime_config.slack_incoming_hook校验sql_client().execute_query验证目标库真实数据last_trace.last_normalize_info.row_counts低成本验证归一化结果并用断言锁死预期行数可观测把load_info、trace自动脱敏、schema 更新明细回写_load_info/_trace/_new_tables让流水线元数据变成可 SQL 查询的数据资产。进一步阅读官方示例脚本全文chess_production.md重试与 schema 更新策略源码dlt/pipeline/helpers.pySlack 通知实现dlt/common/runtime/slack.pyLoadInfo/NormalizeInfo定义dlt/common/pipeline.pypipeline.sql_client与last_trace实现dlt/pipeline/pipeline.py归一化信息与 trace 访问dlt/pipeline/trace.py同系列的 transformer 深入示例transformers.md增量加载场景可参考 incremental_loading.md【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 dlt 构建 Chess API 生产级数据流水线：重试、通知、可观测性与自动化测试实战
用 dlt 构建 Chess API 生产级数据流水线重试、通知、可观测性与自动化测试实战【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt本技术指南基于开源数据加载工具 dltdata load tool官方示例 chess_production.md带你从零构建一条对接 Chess.com 公共 API 的生产级数据流水线。示例脚本展示了如何提取棋手资料与对局数据并把「重试机制、Slack 通知、加载结果检查、模式演进告警、自动化测试、LoadInfo/Trace 回写目标库」等生产环境必备能力全部落地到同一段可运行的代码中。读完本文你将掌握 dlt 中source/transformer组合建模、基于 tenacity 的分阶段重试、sql_client与normalize_info两种数据校验方式以及如何把流水线运行元数据作为数据本身存回目标数据仓库。示例概览这条流水线做了什么该示例是一个完整的 Python 脚本与 Chess.com 公共 APIhttps://api.chess.com/pub/交互抽取棋手列表按头衔如 GM及其个人资料、月度对局数据最终写入 duckdb 数据库。它的特殊之处在于不是为了演示“能跑通”而是演示“上线之后怎么管理”——脚本在生产运行流程中依次完成以下工作加载完成后检查加载包load package逐个查看其中的 job 与 schema 变更把加载信息load info、schema 更新、运行 trace作为数据回写目标库形成可查询的元数据表在**模式演进schema evolution**发生时触发通知使用上下文管理器与 tenacity 重试策略独立重试流水线中的不同阶段借助sql_client与normalize_info编写基础数据校验测试如检查表行数并断言。官方文档给出的学习目标在 chess_production.md 中列出本文按脚本的实际执行顺序逐段拆解。数据源建模source resource transformer 的组合示例的核心数据源定义在dlt.source装饰的chess()函数中chess_production.md它把三个资源组合成一条完整的数据 DAGdlt.source def chess( chess_url: str dlt.config.value, title: str GM, max_players: int 2, year: int 2022, month: int 10, ) - Any: def _get_data_with_retry(path: str) - StrAny: r client.get(f{chess_url}{path}) return r.json()通过dlt.config.value注入外部配置注意第一个参数chess_url: str dlt.config.valuedlt 会从环境变量、TOML 配置等来源注入该值不需要在代码里写死 API 地址。这就是 dlt 的配置注入机制与之配套的title、max_players、year、month都是普通默认参数可在调用时覆盖。并行化 transformerplayers_profilesdlt.transformer(data_fromplayers, write_dispositionreplace, parallelizedTrue) def players_profiles(username: Any) - TDataItems: print(fgetting {username} profile via thread {threading.current_thread().name}) sleep(1) # add some latency to show parallel runs return _get_data_with_retry(fplayer/{username})players资源逐个 yield 棋手用户名yield from ...[players][:max_players]players_profiles作为 transformer 消费这些用户名并抓取各自的 profile。parallelizedTrue让 transformer 在线程池中并行运行代码中的threading.current_thread().name与sleep(1)正是为了在运行时直观展示并行效果这是 dlt 提供的资源级并发能力。容忍缺失数据的 transformerplayers_gamesdlt.transformer(data_fromplayers, write_dispositionappend) def players_games(username: Any) - Iterator[TDataItems]: path fplayer/{username}/games/{year:04d}/{month:02d} try: yield _get_data_with_retry(path)[games] except requests.HTTPError as exc: # we allow players to not have games for some months if not exc.response.status_code 404: raise excplayers_games抓取指定年月的对局但对 404 做了显式容忍——某些棋手在某月确实没有对局记录这不是错误。这一模式在生产中非常常见对上游 API 的业务性缺失与真正的故障要区别对待。三个资源使用了两种写入策略players/players_profiles用replace全量替换players_games用append增量追加配合使用能让数据表各取所需。关于并行化 transformer 与资源编排的更多通用用法可参考 transformers.mddlt 的 source 装饰器与资源系统定义位于 dlt/extract/source.py 与 dlt/extract/resource.py。生产级重试tenacity 与 retry_load 组合拳数据流水线上线后网络抖动、目标库临时不可用都是常态。示例没有把重试逻辑写死在pipeline.run外面套一个简单循环而是使用 tenacity 的Retrying对象包住每次运行尝试并配合 dlt 提供的retry_load判定函数chess_production.mddef load_data_with_retry(pipeline, data): try: for attempt in Retrying( stopstop_after_attempt(5), waitwait_exponential(multiplier1.5, min4, max10), retryretry_if_exception(retry_load(())), reraiseTrue, ): with attempt: logger.info( fRunning the pipeline, attempt{attempt.retry_state.attempt_number} ) load_info pipeline.run(data) logger.info(str(load_info)) send_slack_message( pipeline.runtime_config.slack_incoming_hook, Data was successfully loaded!, ) except Exception: # we get here after all the failed retries send_slack_message( pipeline.runtime_config.slack_incoming_hook, Something went wrong! ) raise这里load_data_with_retry是自定义辅助函数不是 dlt 内置 API它的重试策略参数有明确语义参数值含义stopstop_after_attempt(5)最多尝试 5 次waitwait_exponential(multiplier1.5, min4, max10)指数退避等待基础乘数 1.5等待 4~10 秒retryretry_if_exception(retry_load(()))只有满足retry_load判定条件的异常才重试reraiseTrue重试耗尽后抛出原始异常retry_load 的判定逻辑什么该重试什么不该重试retry_load是 dlt 提供的内置重试策略工厂实现在 dlt/pipeline/helpers.py。其核心逻辑是只重试load阶段如果异常是PipelineStepFailed且出错的步骤不在retry_on_pipeline_steps默认只有load中则不重试——extract、normalize 阶段失败通常不是网络抖动重试没有意义不重试终态异常TerminalException配置缺失、认证失败、job 终态失败等异常重复多少次都不会成功见 dlt/pipeline/helpers.py 对TerminalException的判断。该函数的 docstring 给出了等价的最小用法示例dlt/pipeline/helpers.pyfor attempt in Retrying(stopstop_after_attempt(3), retryretry_if_exception(retry_load(())), reraiseTrue): with attempt: p.run(data)retry_load(())中的空元组表示只重试load步骤本身等价于默认(load,)。如果你希望其他阶段也参与重试可以传入如retry_load((load, extract))如果需要处理“多流水线并行时 schema 竞争创建表/加列失败”的场景dlt 还提供了retry_schema_update()策略可与retry_load用|组合见 dlt/pipeline/helpers.py。加载结果的深度检查LoadInfo 与 load_packages成功重试并完成一次pipeline.run后返回的load_info是LoadInfo类型——dlt 定义的“最近一次加载包信息”元组由pipeline.run和load方法返回dlt/common/pipeline.py。示例用它做了四件事chess_production.md# see when load was started logger.info(fPipeline was started: {load_info.started_at}) # print the information on the first load package and all jobs inside logger.info(fFirst load package info: {load_info.load_packages[0]}) # print the information on the first completed job in first load package logger.info( fFirst completed job info: {load_info.load_packages[0].jobs[completed_jobs][0]} ) # check for schema updates: schema_updates [p.schema_update for p in load_info.load_packages] if schema_updates: send_slack_message( pipeline.runtime_config.slack_incoming_hook, Schema was updated! )要点拆解load_info.started_at/finished_at本次 load 阶段的起止时间LoadInfo.asstr()会自动格式化输出“load 阶段耗时、加载了多少个包”等人类可读摘要dlt/common/pipeline.pyload_info.load_packagesLoadPackageInfo列表定义于 dlt/common/storages/load_package.py每个包对应一次独立的加载批次包内有完整的 job 状态分类jobs[completed_jobs]等package.schema_update本次加载引起的 schema 变更集合。只要非空就说明上游数据让 dlt 自动演进出了新表或新列——这是数据漂移的早期信号示例选择立即向 Slack 发送 “Schema was updated!” 通知。这样就把“数据有没有变化、变化了什么”变成可观测、可告警的信号。用 sql_client 与 normalize_info 做自动化校验生产流水线需要在加载后做基本的数据质量断言。示例提供了两种互补的校验手段chess_production.md方式一sql_client 直接查目标库with pipeline.sql_client() as client: with client.execute_query(SELECT COUNT(*) FROM players) as cursor: count cursor.fetchone()[0] if count 0: logger.info(Warning: No data in players table) else: logger.info(fPlayers table contains {count} rows) assert count MAX_PLAYERSpipeline.sql_client()是 dlt 在 dlt/pipeline/pipeline.py 中提供的标准入口它返回一个已认证且默认限定在流水线 dataset上的 SQL 客户端用with语句管理连接生命周期。其 docstring 给出了通用查询示例含参数化查询with pipeline.sql_client() as client: with client.execute_query( SELECT id, name, email FROM customers WHERE id %s, 10 ) as cursor: print(cursor.fetchall())需要注意的是sql_client只在目标库支持 SQL 接口时可用例如 duckdb、postgres、snowflake 等对于纯文件类目标如 filesystem会抛出SqlClientNotAvailable见 dlt/pipeline/pipeline.py。方式二normalize_info 直接读元数据normalize_info pipeline.last_trace.last_normalize_info count normalize_info.row_counts.get(players, 0) if count 0: logger.info(Warning: No data in players table) else: logger.info(fPlayers table contains {count} rows) assert count MAX_PLAYERSpipeline.last_trace属性从标准位置加载最近一次运行 tracedlt/pipeline/pipeline.pylast_normalize_info从 trace 中取出 normalize 阶段的NormalizeInfodlt/pipeline/trace.py。NormalizeInfo.row_counts属性按表汇总了本次归一化处理的数据项数量dlt/common/pipeline.py。两种方式各有优势sql_client校验的是目标库里的真实数据端到端最可靠normalize_info校验的是归一化阶段的处理结果不需要真正查询目标库成本低、与目标无关。示例把两者都断言为MAX_PLAYERS脚本顶部定义为 5见 chess_production.md这样任何一步丢失数据都会立刻让测试失败。把运行元数据回写目标库load_info、trace 与 schema 更新表示例最具“生产可观测性”特色的一步是把流水线自身的运行元数据当作普通数据加载回同一个目标库chess_production.md# we reuse the pipeline instance below and load to the same dataset as data logger.info(Saving the load info in the destination) pipeline.run([load_info], table_name_load_info) assert _load_info in pipeline.last_trace.last_normalize_info.row_counts # save trace to destination, sensitive data will be removed logger.info(Saving the trace in the destination) pipeline.run([pipeline.last_trace], table_name_trace) assert _trace in pipeline.last_trace.last_normalize_info.row_counts这里有三个关键点复用 pipeline 实例再次调用pipeline.run时数据会加载到与业务数据相同的 dataset 中只是换成了_load_info、_trace这样的下划线前缀表名与业务表区分开LoadInfo/Trace本身就是可加载对象dlt 为它们实现了asdict()序列化LoadInfo.asdict见 dlt/common/pipeline.py会展开为带load_id的job_metrics、outputs明细行所以可以直接塞进pipeline.runtrace 在落库时会自动移除敏感数据文档注释明确说明 “sensitive data will be removed”用 row_counts 断言落库成功_load_info in pipeline.last_trace.last_normalize_info.row_counts确认元数据表确实写入了。接着示例把本次 schema 演进的明细也存了下来chess_production.md# print all the new tables/columns in for package in load_info.load_packages: for table_name, table in package.schema_update.items(): logger.info(fTable {table_name}: {table.get(description)}) for column_name, column in table[columns].items(): logger.info(f\tcolumn {column_name}: {column[data_type]}) # save the new tables and column schemas to the destination: table_updates [p.asdict()[tables] for p in load_info.load_packages] pipeline.run(table_updates, table_name_new_tables) assert _new_tables in pipeline.last_trace.last_normalize_info.row_counts先在日志中打印新增的表名、描述、列名与数据类型再把load_packages的asdict()[tables]结构整体写入_new_tables表。这样模式变更本身就成了可查询的数据——将来排查“哪次运行引入了哪张新表/新列”直接查_new_tables即可配合_load_info、_trace就能还原每次运行的全貌。运行方式与完整脚本脚本入口在 chess_production.mdif __name__ __main__: # create dlt pipeline pipeline dlt.pipeline( pipeline_namechess_pipeline, destinationduckdb, dataset_namechess_data, ) # get data for a few famous players data chess(max_playersMAX_PLAYERS) load_data_with_retry(pipeline, data)运行前提安装 dlt 及 duckdb 目标支持pip install dlt[duckdb]另需requests、tenacity依赖通过dlt.pipeline(pipeline_namechess_pipeline, destinationduckdb, dataset_namechess_data)创建流水线——pipeline_name用于本地状态/trace 存储destination指定目标库类型dataset_name指定目标数据集chess(max_playersMAX_PLAYERS)实例化数据源load_data_with_retry负责带重试地运行并执行全部检查与回写chess_url未在代码中赋值需通过 dlt 配置注入例如设置环境变量或用 TOML 配置例如CHESS__CHESS_URL环境变量对应chess_url参数dlt 按资源名__参数名的层级自动解析。运行后预期观察到的现象与源码逻辑一一对应日志中出现多次 “Running the pipeline, attempt1/2/…”仅在遇到可重试异常时出现多次players_profiles的打印中线程名各不相同并行化生效日志输出LoadInfo摘要、started_at、首个 load package 与其 completed jobs 明细若发生 schema 演进Slack 收到 “Schema was updated!”目标库chess_data数据集中新增players、players_profiles、players_games业务表以及_load_info、_trace、_new_tables三张元数据表assert count MAX_PLAYERS保证 players 表行数与预期严格一致。如果想观察“并行化 逐项处理”的 transformer 行为可以把max_players调大_get_data_with_retry复用了 dlt 内置的带重试 HTTP 客户端dlt.sources.helpers.requests.client见 dlt/sources/helpers/requests它对公共 API 的瞬时故障有额外一层兜底。小结一条生产流水线的完整检查清单把 chess_production.md 中演示的能力整理成清单可直接作为你自己 dlt 流水线的上线模板建模用dlt.sourcedlt.resourcedlt.transformer组合数据 DAG对可并行任务开parallelizedTrue用write_disposition区分replace/append语义对上游 404 等业务性缺失显式容忍配置敏感地址与密钥通过dlt.config.value从环境变量/TOML 注入不写死在代码里重试用 tenacityRetrying包裹运行配合retry_load只重试load阶段、避开终态异常必要时用retry_schema_update应对并发 schema 竞争告警成功、失败、schema 演进三类事件通过send_slack_message通知到pipeline.runtime_config.slack_incoming_hook校验sql_client().execute_query验证目标库真实数据last_trace.last_normalize_info.row_counts低成本验证归一化结果并用断言锁死预期行数可观测把load_info、trace自动脱敏、schema 更新明细回写_load_info/_trace/_new_tables让流水线元数据变成可 SQL 查询的数据资产。进一步阅读官方示例脚本全文chess_production.md重试与 schema 更新策略源码dlt/pipeline/helpers.pySlack 通知实现dlt/common/runtime/slack.pyLoadInfo/NormalizeInfo定义dlt/common/pipeline.pypipeline.sql_client与last_trace实现dlt/pipeline/pipeline.py归一化信息与 trace 访问dlt/pipeline/trace.py同系列的 transformer 深入示例transformers.md增量加载场景可参考 incremental_loading.md【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考