尧图网站建设 
dlt Pipeline State 完全指南状态存储、跨资源共享、目的地同步与重置【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dltdltdata load tool的 Pipeline State 是与数据并排持久化的 Python 字典用于跨管道运行保存和恢复元数据——它既是dlt.sources.incremental实现增量加载的底层基石也是你在自定义资源中记录已处理实体、去重请求、保存游标值的官方推荐手段。读完本文你将掌握dlt.current.resource_state()与dlt.current.source_state()的读写语义、状态与目的地destination的同步机制_dlt_pipeline_state表、dlt pipeline命令族的状态检查与全量/部分重置方法以及何时该放弃 state 转而直接查询已加载数据。什么是 Pipeline State从定义上讲Pipeline State 是一个与你的数据一同存放的 Python 字典你可以在一次管道运行中向它写入值并在下一次管道运行时把它们取回来。它并不额外依赖任何外部服务而是随数据一起被提取、规范化和加载天然支持增量场景。在仓库源码中state 的读写入口集中在 dlt/extract/state.pysource_state(source_state_key)返回源级作用域的 state 字典resource_state(resource_nameNone)返回资源级作用域的 state 字典reset_resource_state(resource_name)/delete_source_state_keys(key)用于按资源或按 JSONPath 键删除 state 片段。resource_state的实际实现会先在源级 state 下维护resources键再按资源名取子字典见 dlt/extract/state.py#L123-L127即state_[resources][resource_name]source_state则直接在源级 state 下按源名取子字典dlt/extract/state.py#L64。所有对字典内容的就地修改如append、setdefault都会在下一次数据加载时被一并持久化。这些函数统一通过dlt.current暴露给业务代码dlt.current模块同时提供pipeline()、source()、resource()等访问器见 dlt/pipeline/current.py。在资源中读写 Pipeline State最典型的用法是在dlt.resource装饰的函数内读写 state。下面的示例出自 state.md 原文档与 advanced-state.md 中的进阶讲解同源使用 state 维护一份已处理的象棋对局归档地址清单从而避免重复请求相同数据dlt.resource(write_dispositionappend) def players_games(chess_url, player, start_monthNone, end_monthNone): # create or request a list of archives from resource-scoped state checked_archives dlt.current.resource_state().setdefault(archives, []) # get a list of archives for a particular player archives _get_players_archives(chess_url, player) # ty: ignore[unresolved-reference] for url in archives: if url in checked_archives: print(fskipping archive {url}) continue else: print(fgetting archive {url}) checked_archives.append(url) # get the filtered archive r requests.get(url) r.raise_for_status() yield r.json().get(games, [])这里通过dlt.current.resource_state().setdefault(archives, [])请求资源级 statechecked_archives列表在archives键下它是私有的、仅对players_games资源可见。任何对列表的append操作都会在本次数据加载完成后随数据一起被写回 state。要点如下state 存储于本地Pipeline state 存放在 pipeline working directory 中默认位于用户主目录~/.dlt/pipelines/pipeline_name。因此不同名称的 pipeline 之间无法共享 state——同一个 state 只属于同名 pipeline两个同名 pipeline 脚本会看到同一个工作目录并共享全部 state。必须是 JSON 可序列化state 中的值必须能被 JSON 序列化。除标准 Python 类型外dlt 额外支持DateTime、Decimal、bytes和UUID源码层面在 dlt/extract/state.py 的 docstring 中明确声明支持 dumps/restores 这几种类型。资源级 state 是隔离的一个资源写入的 state 对其他资源不可见这是 dlt 官方推荐的默认作用域。为什么就地修改即可生效在resource_state的实现中dlt/extract/state.py#L77-L127返回的是 state 字典内部对象的直接引用而不是副本。因此对列表调用append、对字典执行setdefault或按键赋值都直接修改了底层 state。这正是dlt.sources.incremental的实现原理之一——它把最后值写入 state下次运行再读出来作为起始游标。需要显式传资源名的场景resource_state()在不传参时会通过get_current_pipe_name()内部基于contextvars的_CURRENT_PIPE_CONTEXT见 dlt/extract/state.py#L160-L202反查当前资源名。如果你在异步函数或使用defer装饰的函数中请求 state且运行在与主线程不同的线程中dlt 可能无法解析资源名此时需要显式传入资源名dlt.current.resource_state(my_resource)。跨资源共享 State源级作用域dlt.current.source_state()返回源级source-scopedstate它可以在同一个 source 的多个资源之间共享并且在dlt.source装饰的函数内部只读可用源码 docstring 明确指出source 装饰函数中获取的 state 是只读的任何修改都会被丢弃而 resource 装饰函数中获取的 state 是可写的见 dlt/extract/state.py#L33-L46。最常用的场景是在源级 state 中存放自定义字段到展示名称的映射字典。官方验证源verified source中的pipedrive管道即采用此模式指定一个资源作为 state 写入者writer其余资源作为 state 读取者reader从而在并行执行时避免写冲突。import dlt dlt.source def my_source(): # 在源装饰函数内读取源级 state只读修改会被丢弃 mapping dlt.current.source_state().get(field_mapping, {}) ... dlt.resource def resource_a(): # 多个资源共享同一个源级 state state dlt.current.source_state() state[field_mapping] {internal: display} # 写入方 ... dlt.resource def resource_b(): mapping dlt.current.source_state().get(field_mapping, {}) # 读取方 ...建议官方 tip尽量对 source 进行分解decomposition以便在 Airflow 上串行或并行执行具体方法见 性能优化文档中的 source decomposition 小节。source().decompose(strategyscc)会把 source 拆成强连通分量串行加载这些分量是安全且有序的如果只用资源级 state还可以用不同的确定性pipeline_name并行运行各分量。若无法避免分解则指定一个资源作为 state 写入者、其余资源作为 state 读取者——这正是pipedrive管道的做法这样部分资源仍可并行运行。与目的地同步 State_dlt_pipeline_state表与dlt pipeline sync如果你在 Airflow 等调度平台上运行 pipeline每次任务都会获得一个干净的文件系统~/.dlt/pipelines/pipeline_name工作目录总是被删除。此时本地 state 会丢失dlt 如何继续增量加载答案是dlt 会把 state 与所有其他数据一同加载到目的地。当面对一个干净的工作目录时Pipeline.run()会尝试从目的地恢复 state 与 schema。该行为由restore_from_destination配置项控制默认值为True见 dlt/pipeline/configuration.py#L44并且在run方法执行任何数据操作之前同步见 dlt/pipeline/pipeline.py 中with_state_sync/with_schemas_sync装饰器与sync_destination相关注释。远程 state 的识别依据是三元组pipeline 名称pipeline_name目的地位置由 credentials 决定如 duckdb 文件路径、BigQuery 项目等目的地数据集dataset_name。要复用同一份 state就必须使用相同的 pipeline 名称和相同的 destination。从源码看该表名被统一定义为_dlt_pipeline_state见 dlt/common/schema/typing.py#L42 的PIPELINE_STATE_TABLE_NAME表中包含 pipeline 信息、state 所属的 pipeline runload id以及 state 本体state blob。CLI 提供了显式从该表取回 state 的命令dlt pipeline pipeline_name syncsync命令会删除本地 pipeline 工作目录含所有待处理包、未同步的 state 变更和 schema然后从目的地恢复最后一次同步的状态目的地中的数据与 schema 不受影响详见 command-line-interface.md 中dlt pipeline sync一节。如果你删除了 pipeline 正在加载的 dataset执行该命令将得到 state 的完全重置。若 pipeline 没有本地工作目录也可以通过--destination和--dataset-name参数加上.dlt/secrets.toml中的目的地凭据从目的地重建一个工作目录。提示如果每次运行之间你都能保留 pipeline 工作目录可以在config.toml中设置restore_from_destinationfalse来禁用 state 同步省去每次恢复的开销。何时使用 Pipeline State官方给出了明确的适用场景清单增量加载的 last value覆盖约 90% 的需求dlt 内部使用 state 实现 last value 增量加载。dlt.sources.incremental把最新游标写入 state并在下次运行通过start_value取回。绝大多数情况下你只需要声明dlt.sources.incremental而不必直接操作 state。存储已请求实体列表详见 advanced-state.md当列表规模不超过约10 万个元素时可以把它存在资源级 state 中用于去重。存储大字典的 last values当标准incremental构造无法满足例如需要为 Twitter API 的每个搜索词分别跟踪最后值时可以在 state 中维护搜索词 → 最后值字典。存储自定义字段字典、动态配置及其他源级 state。作为参照advanced-state.md 给出了两个直接操作 state 的典型模式保存最后时间戳dlt.resource() def tweets(): # Get the last value from loaded metadata. If it does not exist, get None last_val dlt.current.resource_state().setdefault(last_updated, None) # Get data and yield it data _get_data(start_fromlast_val) # ty: ignore[unresolved-reference] yield data # Change the state to the new value dlt.current.resource_state()[last_updated] data[last_timestamp]按搜索词分别跟踪最后值dlt.resource(write_dispositionappend) def search_tweets(twitter_bearer_tokendlt.secrets.value, search_termsNone, start_timeNone, end_timeNone, last_valueNone): for search_term in search_terms: # Make cache for each term last_value_cache dlt.current.resource_state().setdefault(flast_value_{search_term}, None) ... last_id page.get(meta, {}).get(newest_id, 0) # Set it back dlt.current.resource_state()[flast_value_{search_term}] max(last_value_cache or 0, int(last_id)) yield page不要用 Pipeline State 的场景可能膨胀到百万级记录当 state 可能增长到百万个元素时不要使用 dlt state。例如为你的全部数百万条用户记录各保存一条修改时间戳——这几乎肯定是个坏主意。此时可以把 state 存到外部存储DynamoDB、Redis 等。但要注意如果 extract 阶段失败你会得到一个无效的 state需要自行保证写入的原子性与一致性。把已加载的数据当作 state 使用dlt 通过dlt.current.pipeline()暴露当前 pipeline从中可以获取 SQL client见 dlt 生态中的 SQL 访问文档来查询感兴趣的数据。这种情况下至少尽量以批量方式处理用户记录。访问目的地数据替代 Pipeline State官方推荐的替代方案是直接在资源内查询目的地中已有的数据。下面的示例出自原文档加载指定user_id的最新评论通过查询user_comments表取得该用户的最大评论 id再只拉取 id 更大的新评论。import dlt dlt.resource(nameuser_comments) def comments(user_id: str): current_pipeline dlt.current.pipeline() # find the last comment id for the given user_id by looking in the destination max_id: int 0 # on the first pipeline run, the user_comments table does not yet exist so do not check at all # alternatively, catch DatabaseUndefinedRelation which is raised when an unknown table is selected if not current_pipeline.first_run: # get user comments table from pipeline dataset # get last user comment id with ibis expression, ibis-extras need to be installed dataset current_pipeline.dataset() user_comments dataset.table(user_comments).to_ibis() max_id_expression user_comments.filter(user_comments.user_id user_id).select(user_comments[_id].max()) max_id_df dataset(max_id_expression).df() # if there are no comments for the user, max_id will be None, so we replace it with 0 max_id max_id_df[0][0] if len(max_id_df.index) else 0 # ty: ignore # use max_id to filter our results (we simulate an API query) yield from [ {_id: i, value: letter, user_id: user_id} for i, letter in zip([1, 2, 3], [A, B, C]) if i max_id ]这段代码的两个关键处理首次运行跳过目的地查询pipeline 首次运行时目的地数据集和user_comments表尚不存在。这里用 pipeline 的first_run属性跳过查询也可以捕获DatabaseUndefinedRelation异常查询不存在的表时抛出。空结果兜底当某用户还没有评论时max_id为None代码将其替换为0作为初始值。这种以目的地为 state的模式完全绕开了 state 的大小限制——数据表可以容纳任意规模的记录且天然具备容错能力。检查 Pipeline State你可以用dlt pipeline命令族检查 pipeline 的当前状态详见 command-line-interface.md 中dlt pipeline一节dlt pipeline -v chess_pipeline infodlt pipeline info会显示 pipeline 工作目录的内容数据集名称、目的地、schema 列表、schema 中的资源、已完成/待处理的 load package以及在-v详细模式下所有已知 source 的源级与资源级 state 槽位。命令输出中会直接呈现你在资源中写入的 state 键与值便于核对增量游标或去重列表是否正确落盘。重置 Pipeline State全量与部分完全重置有三种方式可以彻底重置 pipeline state删除目的地数据集——数据、schema、state 一并清除pipeline 完全从头开始创建 pipeline 时设置dev_modeTrue——每次运行都从零开始并加载到独立数据集旧 state 与工作目录不再复用配置项见 dlt/pipeline/configuration.py#L52更多背景见 pipeline.md 中 dev_mode 一节使用 CLI 命令drop 命令参考dlt pipeline pipeline_name drop --drop-all该命令会删除指定 schema 的 state 与全部数据表含嵌套表并打印将要执行的修改清单供确认。部分重置如果你只想重置某个资源或 state 中的某个键而不触碰其他数据按资源重置同时删除该资源生成的表并重置其 statedlt pipeline pipeline_name drop resource_name例如 GitHub 管道中重置repo_events资源以强制其全量刷新dlt pipeline github_events drop repo_events。命令支持正则用re:前缀标识例如dlt pipeline github_events drop re:^repo。仅重置指定 state 路径不触碰表和数据dlt pipeline pipeline_name drop --state-paths archives--state-paths接受相对于源级 state 的键名或 JSONPath。例如它会选中chess源 state 下的archives键并删除之{ sources: { chess: { archives: [ https://api.chess.com/pub/player/magnuscarlsen/games/2022/05 ] } } }从源码看这些能力统一由 dlt/pipeline/drop.py 的prepare_drop_resources实现drop_allTrue时编译re:.*匹配全部资源state_paths通过 JSONPath 解析后在源级 state 上执行delete_source_state_keys而state_only模式则只改 state、不改 schema 与目的地表见 dlt/pipeline/drop.py#L82-L167。命令执行前会以 dry-run 方式列出将删除的表、将被重置的资源 state 槽位与 state 路径经确认后才真正生效。注意该 drop 命令目前仍标记为experimental接口可能在未来版本调整。小结选择正确的状态管理策略综合原文档与源码实现可以把 dlt 的状态管理决策简化为一张检查表场景推荐方案依据增量加载最后值绝大多数情况dlt.sources.incremental内部使用 statecursor.md去重已请求实体列表 ≤ 10 万资源级 state 存列表advanced-state.md跨资源共享映射/配置源级 state指定唯一 writerdlt/extract/state.py记录可能达百万级目的地数据或外部存储DynamoDB/Redis本文不要用 state一节调度平台干净文件系统依赖目的地恢复必要时dlt pipeline sync本文与目的地同步一节需要清空/部分清空 statedrop命令族 /dev_modedlt/pipeline/drop.py正确选择 state 作用域与规模边界是让 dlt 管道既保持增量高效、又不被状态膨胀拖垮的关键。建议优先使用dlt.sources.incremental把直接操作 state 留给确实需要自定义跟踪的场景并始终记得state 会随数据一起加载到目的地因此它必须是 JSON 可序列化的且应保持小而精。【免费下载链接】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 Pipeline State 完全指南：状态存储、跨资源共享、目的地同步与重置
dlt Pipeline State 完全指南状态存储、跨资源共享、目的地同步与重置【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dltdltdata load tool的 Pipeline State 是与数据并排持久化的 Python 字典用于跨管道运行保存和恢复元数据——它既是dlt.sources.incremental实现增量加载的底层基石也是你在自定义资源中记录已处理实体、去重请求、保存游标值的官方推荐手段。读完本文你将掌握dlt.current.resource_state()与dlt.current.source_state()的读写语义、状态与目的地destination的同步机制_dlt_pipeline_state表、dlt pipeline命令族的状态检查与全量/部分重置方法以及何时该放弃 state 转而直接查询已加载数据。什么是 Pipeline State从定义上讲Pipeline State 是一个与你的数据一同存放的 Python 字典你可以在一次管道运行中向它写入值并在下一次管道运行时把它们取回来。它并不额外依赖任何外部服务而是随数据一起被提取、规范化和加载天然支持增量场景。在仓库源码中state 的读写入口集中在 dlt/extract/state.pysource_state(source_state_key)返回源级作用域的 state 字典resource_state(resource_nameNone)返回资源级作用域的 state 字典reset_resource_state(resource_name)/delete_source_state_keys(key)用于按资源或按 JSONPath 键删除 state 片段。resource_state的实际实现会先在源级 state 下维护resources键再按资源名取子字典见 dlt/extract/state.py#L123-L127即state_[resources][resource_name]source_state则直接在源级 state 下按源名取子字典dlt/extract/state.py#L64。所有对字典内容的就地修改如append、setdefault都会在下一次数据加载时被一并持久化。这些函数统一通过dlt.current暴露给业务代码dlt.current模块同时提供pipeline()、source()、resource()等访问器见 dlt/pipeline/current.py。在资源中读写 Pipeline State最典型的用法是在dlt.resource装饰的函数内读写 state。下面的示例出自 state.md 原文档与 advanced-state.md 中的进阶讲解同源使用 state 维护一份已处理的象棋对局归档地址清单从而避免重复请求相同数据dlt.resource(write_dispositionappend) def players_games(chess_url, player, start_monthNone, end_monthNone): # create or request a list of archives from resource-scoped state checked_archives dlt.current.resource_state().setdefault(archives, []) # get a list of archives for a particular player archives _get_players_archives(chess_url, player) # ty: ignore[unresolved-reference] for url in archives: if url in checked_archives: print(fskipping archive {url}) continue else: print(fgetting archive {url}) checked_archives.append(url) # get the filtered archive r requests.get(url) r.raise_for_status() yield r.json().get(games, [])这里通过dlt.current.resource_state().setdefault(archives, [])请求资源级 statechecked_archives列表在archives键下它是私有的、仅对players_games资源可见。任何对列表的append操作都会在本次数据加载完成后随数据一起被写回 state。要点如下state 存储于本地Pipeline state 存放在 pipeline working directory 中默认位于用户主目录~/.dlt/pipelines/pipeline_name。因此不同名称的 pipeline 之间无法共享 state——同一个 state 只属于同名 pipeline两个同名 pipeline 脚本会看到同一个工作目录并共享全部 state。必须是 JSON 可序列化state 中的值必须能被 JSON 序列化。除标准 Python 类型外dlt 额外支持DateTime、Decimal、bytes和UUID源码层面在 dlt/extract/state.py 的 docstring 中明确声明支持 dumps/restores 这几种类型。资源级 state 是隔离的一个资源写入的 state 对其他资源不可见这是 dlt 官方推荐的默认作用域。为什么就地修改即可生效在resource_state的实现中dlt/extract/state.py#L77-L127返回的是 state 字典内部对象的直接引用而不是副本。因此对列表调用append、对字典执行setdefault或按键赋值都直接修改了底层 state。这正是dlt.sources.incremental的实现原理之一——它把最后值写入 state下次运行再读出来作为起始游标。需要显式传资源名的场景resource_state()在不传参时会通过get_current_pipe_name()内部基于contextvars的_CURRENT_PIPE_CONTEXT见 dlt/extract/state.py#L160-L202反查当前资源名。如果你在异步函数或使用defer装饰的函数中请求 state且运行在与主线程不同的线程中dlt 可能无法解析资源名此时需要显式传入资源名dlt.current.resource_state(my_resource)。跨资源共享 State源级作用域dlt.current.source_state()返回源级source-scopedstate它可以在同一个 source 的多个资源之间共享并且在dlt.source装饰的函数内部只读可用源码 docstring 明确指出source 装饰函数中获取的 state 是只读的任何修改都会被丢弃而 resource 装饰函数中获取的 state 是可写的见 dlt/extract/state.py#L33-L46。最常用的场景是在源级 state 中存放自定义字段到展示名称的映射字典。官方验证源verified source中的pipedrive管道即采用此模式指定一个资源作为 state 写入者writer其余资源作为 state 读取者reader从而在并行执行时避免写冲突。import dlt dlt.source def my_source(): # 在源装饰函数内读取源级 state只读修改会被丢弃 mapping dlt.current.source_state().get(field_mapping, {}) ... dlt.resource def resource_a(): # 多个资源共享同一个源级 state state dlt.current.source_state() state[field_mapping] {internal: display} # 写入方 ... dlt.resource def resource_b(): mapping dlt.current.source_state().get(field_mapping, {}) # 读取方 ...建议官方 tip尽量对 source 进行分解decomposition以便在 Airflow 上串行或并行执行具体方法见 性能优化文档中的 source decomposition 小节。source().decompose(strategyscc)会把 source 拆成强连通分量串行加载这些分量是安全且有序的如果只用资源级 state还可以用不同的确定性pipeline_name并行运行各分量。若无法避免分解则指定一个资源作为 state 写入者、其余资源作为 state 读取者——这正是pipedrive管道的做法这样部分资源仍可并行运行。与目的地同步 State_dlt_pipeline_state表与dlt pipeline sync如果你在 Airflow 等调度平台上运行 pipeline每次任务都会获得一个干净的文件系统~/.dlt/pipelines/pipeline_name工作目录总是被删除。此时本地 state 会丢失dlt 如何继续增量加载答案是dlt 会把 state 与所有其他数据一同加载到目的地。当面对一个干净的工作目录时Pipeline.run()会尝试从目的地恢复 state 与 schema。该行为由restore_from_destination配置项控制默认值为True见 dlt/pipeline/configuration.py#L44并且在run方法执行任何数据操作之前同步见 dlt/pipeline/pipeline.py 中with_state_sync/with_schemas_sync装饰器与sync_destination相关注释。远程 state 的识别依据是三元组pipeline 名称pipeline_name目的地位置由 credentials 决定如 duckdb 文件路径、BigQuery 项目等目的地数据集dataset_name。要复用同一份 state就必须使用相同的 pipeline 名称和相同的 destination。从源码看该表名被统一定义为_dlt_pipeline_state见 dlt/common/schema/typing.py#L42 的PIPELINE_STATE_TABLE_NAME表中包含 pipeline 信息、state 所属的 pipeline runload id以及 state 本体state blob。CLI 提供了显式从该表取回 state 的命令dlt pipeline pipeline_name syncsync命令会删除本地 pipeline 工作目录含所有待处理包、未同步的 state 变更和 schema然后从目的地恢复最后一次同步的状态目的地中的数据与 schema 不受影响详见 command-line-interface.md 中dlt pipeline sync一节。如果你删除了 pipeline 正在加载的 dataset执行该命令将得到 state 的完全重置。若 pipeline 没有本地工作目录也可以通过--destination和--dataset-name参数加上.dlt/secrets.toml中的目的地凭据从目的地重建一个工作目录。提示如果每次运行之间你都能保留 pipeline 工作目录可以在config.toml中设置restore_from_destinationfalse来禁用 state 同步省去每次恢复的开销。何时使用 Pipeline State官方给出了明确的适用场景清单增量加载的 last value覆盖约 90% 的需求dlt 内部使用 state 实现 last value 增量加载。dlt.sources.incremental把最新游标写入 state并在下次运行通过start_value取回。绝大多数情况下你只需要声明dlt.sources.incremental而不必直接操作 state。存储已请求实体列表详见 advanced-state.md当列表规模不超过约10 万个元素时可以把它存在资源级 state 中用于去重。存储大字典的 last values当标准incremental构造无法满足例如需要为 Twitter API 的每个搜索词分别跟踪最后值时可以在 state 中维护搜索词 → 最后值字典。存储自定义字段字典、动态配置及其他源级 state。作为参照advanced-state.md 给出了两个直接操作 state 的典型模式保存最后时间戳dlt.resource() def tweets(): # Get the last value from loaded metadata. If it does not exist, get None last_val dlt.current.resource_state().setdefault(last_updated, None) # Get data and yield it data _get_data(start_fromlast_val) # ty: ignore[unresolved-reference] yield data # Change the state to the new value dlt.current.resource_state()[last_updated] data[last_timestamp]按搜索词分别跟踪最后值dlt.resource(write_dispositionappend) def search_tweets(twitter_bearer_tokendlt.secrets.value, search_termsNone, start_timeNone, end_timeNone, last_valueNone): for search_term in search_terms: # Make cache for each term last_value_cache dlt.current.resource_state().setdefault(flast_value_{search_term}, None) ... last_id page.get(meta, {}).get(newest_id, 0) # Set it back dlt.current.resource_state()[flast_value_{search_term}] max(last_value_cache or 0, int(last_id)) yield page不要用 Pipeline State 的场景可能膨胀到百万级记录当 state 可能增长到百万个元素时不要使用 dlt state。例如为你的全部数百万条用户记录各保存一条修改时间戳——这几乎肯定是个坏主意。此时可以把 state 存到外部存储DynamoDB、Redis 等。但要注意如果 extract 阶段失败你会得到一个无效的 state需要自行保证写入的原子性与一致性。把已加载的数据当作 state 使用dlt 通过dlt.current.pipeline()暴露当前 pipeline从中可以获取 SQL client见 dlt 生态中的 SQL 访问文档来查询感兴趣的数据。这种情况下至少尽量以批量方式处理用户记录。访问目的地数据替代 Pipeline State官方推荐的替代方案是直接在资源内查询目的地中已有的数据。下面的示例出自原文档加载指定user_id的最新评论通过查询user_comments表取得该用户的最大评论 id再只拉取 id 更大的新评论。import dlt dlt.resource(nameuser_comments) def comments(user_id: str): current_pipeline dlt.current.pipeline() # find the last comment id for the given user_id by looking in the destination max_id: int 0 # on the first pipeline run, the user_comments table does not yet exist so do not check at all # alternatively, catch DatabaseUndefinedRelation which is raised when an unknown table is selected if not current_pipeline.first_run: # get user comments table from pipeline dataset # get last user comment id with ibis expression, ibis-extras need to be installed dataset current_pipeline.dataset() user_comments dataset.table(user_comments).to_ibis() max_id_expression user_comments.filter(user_comments.user_id user_id).select(user_comments[_id].max()) max_id_df dataset(max_id_expression).df() # if there are no comments for the user, max_id will be None, so we replace it with 0 max_id max_id_df[0][0] if len(max_id_df.index) else 0 # ty: ignore # use max_id to filter our results (we simulate an API query) yield from [ {_id: i, value: letter, user_id: user_id} for i, letter in zip([1, 2, 3], [A, B, C]) if i max_id ]这段代码的两个关键处理首次运行跳过目的地查询pipeline 首次运行时目的地数据集和user_comments表尚不存在。这里用 pipeline 的first_run属性跳过查询也可以捕获DatabaseUndefinedRelation异常查询不存在的表时抛出。空结果兜底当某用户还没有评论时max_id为None代码将其替换为0作为初始值。这种以目的地为 state的模式完全绕开了 state 的大小限制——数据表可以容纳任意规模的记录且天然具备容错能力。检查 Pipeline State你可以用dlt pipeline命令族检查 pipeline 的当前状态详见 command-line-interface.md 中dlt pipeline一节dlt pipeline -v chess_pipeline infodlt pipeline info会显示 pipeline 工作目录的内容数据集名称、目的地、schema 列表、schema 中的资源、已完成/待处理的 load package以及在-v详细模式下所有已知 source 的源级与资源级 state 槽位。命令输出中会直接呈现你在资源中写入的 state 键与值便于核对增量游标或去重列表是否正确落盘。重置 Pipeline State全量与部分完全重置有三种方式可以彻底重置 pipeline state删除目的地数据集——数据、schema、state 一并清除pipeline 完全从头开始创建 pipeline 时设置dev_modeTrue——每次运行都从零开始并加载到独立数据集旧 state 与工作目录不再复用配置项见 dlt/pipeline/configuration.py#L52更多背景见 pipeline.md 中 dev_mode 一节使用 CLI 命令drop 命令参考dlt pipeline pipeline_name drop --drop-all该命令会删除指定 schema 的 state 与全部数据表含嵌套表并打印将要执行的修改清单供确认。部分重置如果你只想重置某个资源或 state 中的某个键而不触碰其他数据按资源重置同时删除该资源生成的表并重置其 statedlt pipeline pipeline_name drop resource_name例如 GitHub 管道中重置repo_events资源以强制其全量刷新dlt pipeline github_events drop repo_events。命令支持正则用re:前缀标识例如dlt pipeline github_events drop re:^repo。仅重置指定 state 路径不触碰表和数据dlt pipeline pipeline_name drop --state-paths archives--state-paths接受相对于源级 state 的键名或 JSONPath。例如它会选中chess源 state 下的archives键并删除之{ sources: { chess: { archives: [ https://api.chess.com/pub/player/magnuscarlsen/games/2022/05 ] } } }从源码看这些能力统一由 dlt/pipeline/drop.py 的prepare_drop_resources实现drop_allTrue时编译re:.*匹配全部资源state_paths通过 JSONPath 解析后在源级 state 上执行delete_source_state_keys而state_only模式则只改 state、不改 schema 与目的地表见 dlt/pipeline/drop.py#L82-L167。命令执行前会以 dry-run 方式列出将删除的表、将被重置的资源 state 槽位与 state 路径经确认后才真正生效。注意该 drop 命令目前仍标记为experimental接口可能在未来版本调整。小结选择正确的状态管理策略综合原文档与源码实现可以把 dlt 的状态管理决策简化为一张检查表场景推荐方案依据增量加载最后值绝大多数情况dlt.sources.incremental内部使用 statecursor.md去重已请求实体列表 ≤ 10 万资源级 state 存列表advanced-state.md跨资源共享映射/配置源级 state指定唯一 writerdlt/extract/state.py记录可能达百万级目的地数据或外部存储DynamoDB/Redis本文不要用 state一节调度平台干净文件系统依赖目的地恢复必要时dlt pipeline sync本文与目的地同步一节需要清空/部分清空 statedrop命令族 /dev_modedlt/pipeline/drop.py正确选择 state 作用域与规模边界是让 dlt 管道既保持增量高效、又不被状态膨胀拖垮的关键。建议优先使用dlt.sources.incremental把直接操作 state 留给确实需要自定义跟踪的场景并始终记得state 会随数据一起加载到目的地因此它必须是 JSON 可序列化的且应保持小而精。【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考