dlt 1.19 版本亮点Arrow 流式查询、Parquet 加速入库、Schema 可视化与 Snowflake 聚类键增强【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt版本导览dltdata load tool1.19 版本围绕「大数据量查询的增量处理」「SQL 数据库加载性能」「Schema 可视化」和「Snowflake 大型表运维」四个方向带来了一系列实用更新。本指南以官方 Release 亮点文档为基础逐项剖析新特性的使用方式、参数细节与底层实现逻辑并给出与仓库源码对应的验证路径帮助你快速判断哪些能力可以直接接入现有数据管道。一、ConnectorX 后端支持 Arrow 流式返回特性概述dlt 内置的sql_database源源码位于 dlt/sources/sql_database/init.py在 1.19 之前默认通过 SQLAlchemy 逐批拉取数据。本次新增的能力是当使用ConnectorX后端时可以将查询结果以Arrow 流Arrow stream形式返回实现大结果集的增量处理避免将全部数据一次性载入内存。启用方式arrow_stream是 ConnectorX 后端的可选返回类型需要显式开启因为 ConnectorX 的默认返回类型是完整的 PyArrow 表from dlt.sources.sql_database import sql_database db sql_database( backendconnectorx, backend_kwargs{ return_type: arrow_stream, # new in 1.19 }, )底层实现原理从源码可以看到ConnectorX 的后端加载器为ConnectorXTableLoader定义于 dlt/sources/sql_database/helpers.py其_load_rows_connectorx方法对return_type做了显式分支处理未指定return_type时默认设置为arrow即一次性返回完整的 PyArrow 表指定为arrow_stream时进入流式模式通过cx.read_sql(conn, query_str, **backend_kwargs)拿到RecordBatchReader逐批迭代record_batch并转换为pa.Table后yield给下游流式模式下会自动把batch_size默认设置为当前资源的chunk_size默认 50000 行即每个批次的行数上限与sql_database/sql_table的chunk_size参数联动。is_streaming False if return_type in backend_kwargs: if backend_kwargs[return_type] arrow_stream: is_streaming True backend_kwargs[batch_size] backend_kwargs.get(batch_size, self.chunk_size) else: backend_kwargs[return_type] arrow此外ConnectorX 默认使用protocol: binary传输协议返回的数据还会经过cast_connectorx_temporal_columns与_maybe_fix_0000_timezone处理用于修正时区与特殊时间值保证时序数据在后续目标 Schema 中的正确性。使用要点arrow_stream属于 ConnectorX 后端专属配置其他后端sqlalchemy、pyarrow、pandas不适用流式模式适合超大表的全量抽取如果数据集规模小于chunk_size非流式的arrow模式依然足够ConnectorX 不依赖 SQLAlchemy 的行级缓冲速度通常更快但需要安装connectorx依赖未安装时会抛出MissingDependencyException该能力同样适用于单表资源sql_table例如结合增量字段加载from dlt.sources.sql_database import sql_table table sql_table( tableorders, backendconnectorx, backend_kwargs{return_type: arrow_stream, batch_size: 100000}, incrementaldlt.sources.incremental(updated_at), )二、Dashboard 新增可视化 Pipeline 运行历史特性概述dlt 的本地 Dashboard 在 1.19 中新增了pipeline 运行历史的可视化视图用于直观展示每次运行的状态、耗时与随时间的变化趋势帮助数据工程师快速判断 pipeline 健康状况、定位失败任务。使用方式在本地运行 pipeline 之后通过 CLI 打开 Dashboard 即可查看运行历史图表dlt pipeline pipeline_name dashboard注意运行历史可视化依赖 dlt 运行期间生成的 trace 与状态数据参见 dlt/pipeline/trace.py 与 dlt/pipeline/state_sync.py因此只有实际执行过pipeline.run()的 pipeline 才会展示出有意义的运行记录。应用价值失败诊断加速从时间轴上快速定位失败运行点再结合日志深入排查运行周期观察查看每次运行的时长波动识别慢查询或资源瓶颈健康度巡检周期性检查运行状态分布及时发现间歇性失败。三、通过 ADBC 将 Parquet 快速导入 MSSQL / MySQL / SQLite特性概述1.19 起dlt 可以使用ADBCArrow Database Connectivity驱动将Parquet 文件直接导入 SQL 数据库MSSQL、MySQL、SQLite。当检测到可用的 ADBC 驱动时Parquet 加载会被自动启用并作为首选方式相比基于INSERT的行级写入可带来10 倍到 100 倍的加载速度提升且比 CSV 回退方案更可靠。自动启用与手动回退由于是自动探测驱动的机制你无需修改任何配置即可在支持的环境中享受加速。如果需要显式退回传统的INSERT加载可以在pipeline.run中指定loader_file_formatinsert_valuespipeline.run( iter(()), # your data generator loader_file_formatinsert_values, )工作机制ADBC 走的是「列式批处理 文件级导入」的通道数据先以 Parquet 文件落盘再由 ADBC 驱动以批量列式方式写入目标库绕开了逐行构造INSERT语句的开销。从加载器实现看文件格式的解析与写入任务由 dlt/load/load.py 与 dlt/destinations/job_client_impl.py 中对应的 SQL 目标实现协同完成ADBC 驱动的可用性决定了自动启用与否。适用前提与限制目标库限定为MSSQL、MySQL、SQLite三类 SQL 数据库环境必须存在对应的 ADBC 驱动例如adbc-driver-sqlite、adbc-driver-postgresql等驱动缺失时 dlt 会回退到原有加载路径若你的业务对写入语句有特殊要求例如完全自定义的 insert 逻辑可显式指定insert_values格式关闭自动加速。四、用Schema.to_mermaid()可视化任意 dlt Schema特性概述1.19 为Schema对象新增了to_mermaid()方法可将任意 dlt Schema 导出为Mermaid ER 图适用于文档编写、Pull Request 评审说明、新成员上手材料等场景。导出结果可直接在 GitHub Markdown、Notion 等支持 Mermaid 渲染的工具中原生展示。核心 APIschema_mermaid pipeline.default_schema.to_mermaid()to_mermaid()定义于 dlt/common/schema/schema.py其内部将 Schema 序列化后委托给 dlt/helpers/mermaid.py 中的schema_to_mermaid生成 Mermaid 字符串。方法签名如下def to_mermaid( self, remove_processing_hints: bool False, hide_columns: bool False, hide_descriptions: bool False, include_dlt_tables: bool True, ) - str:参数默认值作用remove_processing_hintsFalse移除数据处理类 hints如x-normalizer、x-loader标记与冗余信息缩小 Schema 体积、提升可读性hide_columnsFalse隐藏列细节适合表数量多、图幅过大的场景hide_descriptionsFalse隐藏列描述文本include_dlt_tablesTrue是否包含内部 dlt 表_dlt_version、_dlt_loads、_dlt_pipeline_state完整示例import dlt pipeline dlt.pipeline(pipeline_nameevents, destinationduckdb) dlt.resource def events(): yield {event_id: 1, country: DE, ts: 2024-01-01T00:00:00Z} pipeline.run(events()) # 导出为 Mermaid 字符串 print(pipeline.default_schema.to_mermaid()) # 精简模式去掉处理 hints、隐藏描述与内部表 mermaid_clean pipeline.default_schema.to_mermaid( remove_processing_hintsTrue, hide_descriptionsTrue, include_dlt_tablesFalse, )命令行导出除 Python API 外也可以在 CLI 中导出 Schemadlt的 CLI 工具链内部同样使用了to_mermaid参见 dlt/_workspace/cli/utils.py适合在 CI 或文档构建流程中自动化生成 Schema 图。五、Snowflake 聚类键Clustering Key增强特性概述Snowflake 目标在 1.19 中开始支持基于列 hints 更新聚类键。聚类变更会在触发表结构变更时例如新增列自动应用从而无需重建大表即可持续调优聚类策略对超大表的查询性能维护非常友好。使用方法通过apply_hints给列打上cluster: True标记即可dlt.resource(table_nameevents) def events(): yield {event_id: 1, country: DE} events.apply_hints(columns[{name: event_id, cluster: True}]) pipeline.run(events())底层实现逻辑聚类 SQL 的生成位于 dlt/destinations/impl/snowflake/snowflake.py_get_cluster_sql根据列名列表生成CLUSTER BY (col1, col2, ...)语句列名会经escape_column_name转义_get_alter_cluster_sql生成独立的ALTER TABLE ... CLUSTER BY (...)语句_add_cluster_sql区分两种场景建表阶段将CLUSTER BY追加进CREATE TABLE语句而表已存在、需要变更时generate_alterTrue则签发单独的ALTER TABLE语句执行聚类键更新避免整表重建。由此可以确认聚类键的更新发生在 dlt 判定需要变更表结构如新增列触发ALTER时自动随行执行不需要手动重建表。使用要点一个表可以有多个聚类列全部通过cluster: True标记即可聚类键变更属于结构变更操作会随下一次表结构变更触发因此如果只是数据量增长而不涉及结构变化需结合 dlt 的表结构演进机制例如新增字段来触发该能力面向大型事实表、事件表的查询性能优化具体收益取决于查询模式与数据分布。六、其他更新与社区贡献感谢新贡献者1.19 版本收到了来自社区的积极贡献包括 zjacom、JayJai04、anair123、martinibach、hello-world-bfree、tahamuzammil100、timH6502、wrussell1999、luqmansen 等开发者对应 PR 编号 #3294 至 #3332涉及 bug 修复与功能改进。完整的变更列表完整的 1.19 变更清单可参考项目 Release 记录本文聚焦于四项核心用户可见特性特性核心价值相关源码ConnectorX Arrow 流式大结果集增量处理控制内存dlt/sources/sql_database/helpers.pyDashboard 运行历史可视化运行状态与耗时趋势dlt/pipeline/trace.pyADBC 导入 Parquet10–100 倍 SQL 库加载提速dlt/load/load.pySchema.to_mermaid()Schema 一键导出为 Mermaid 图dlt/common/schema/schema.pySnowflake 聚类键更新免重建大表、持续调优聚类dlt/destinations/impl/snowflake/snowflake.py升级建议如果你的sql_database管道经常抽取超大表优先尝试backendconnectorxreturn_typearrow_stream并结合batch_size控制内存水位如果目标库是 MSSQL / MySQL / SQLite 且加载吞吐是瓶颈检查环境是否可安装对应 ADBC 驱动以自动获得 Parquet 批量加载加速在文档与代码评审中推广to_mermaid()把 Schema 演进可视化纳入团队协作流程对 Snowflake 大表用户通过apply_hints(columns[...{cluster: True}])建立聚类策略减少手动 DDL 维护成本。【免费下载链接】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 1.19 版本亮点：Arrow 流式查询、Parquet 加速入库、Schema 可视化与 Snowflake 聚类键增强
dlt 1.19 版本亮点Arrow 流式查询、Parquet 加速入库、Schema 可视化与 Snowflake 聚类键增强【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt版本导览dltdata load tool1.19 版本围绕「大数据量查询的增量处理」「SQL 数据库加载性能」「Schema 可视化」和「Snowflake 大型表运维」四个方向带来了一系列实用更新。本指南以官方 Release 亮点文档为基础逐项剖析新特性的使用方式、参数细节与底层实现逻辑并给出与仓库源码对应的验证路径帮助你快速判断哪些能力可以直接接入现有数据管道。一、ConnectorX 后端支持 Arrow 流式返回特性概述dlt 内置的sql_database源源码位于 dlt/sources/sql_database/init.py在 1.19 之前默认通过 SQLAlchemy 逐批拉取数据。本次新增的能力是当使用ConnectorX后端时可以将查询结果以Arrow 流Arrow stream形式返回实现大结果集的增量处理避免将全部数据一次性载入内存。启用方式arrow_stream是 ConnectorX 后端的可选返回类型需要显式开启因为 ConnectorX 的默认返回类型是完整的 PyArrow 表from dlt.sources.sql_database import sql_database db sql_database( backendconnectorx, backend_kwargs{ return_type: arrow_stream, # new in 1.19 }, )底层实现原理从源码可以看到ConnectorX 的后端加载器为ConnectorXTableLoader定义于 dlt/sources/sql_database/helpers.py其_load_rows_connectorx方法对return_type做了显式分支处理未指定return_type时默认设置为arrow即一次性返回完整的 PyArrow 表指定为arrow_stream时进入流式模式通过cx.read_sql(conn, query_str, **backend_kwargs)拿到RecordBatchReader逐批迭代record_batch并转换为pa.Table后yield给下游流式模式下会自动把batch_size默认设置为当前资源的chunk_size默认 50000 行即每个批次的行数上限与sql_database/sql_table的chunk_size参数联动。is_streaming False if return_type in backend_kwargs: if backend_kwargs[return_type] arrow_stream: is_streaming True backend_kwargs[batch_size] backend_kwargs.get(batch_size, self.chunk_size) else: backend_kwargs[return_type] arrow此外ConnectorX 默认使用protocol: binary传输协议返回的数据还会经过cast_connectorx_temporal_columns与_maybe_fix_0000_timezone处理用于修正时区与特殊时间值保证时序数据在后续目标 Schema 中的正确性。使用要点arrow_stream属于 ConnectorX 后端专属配置其他后端sqlalchemy、pyarrow、pandas不适用流式模式适合超大表的全量抽取如果数据集规模小于chunk_size非流式的arrow模式依然足够ConnectorX 不依赖 SQLAlchemy 的行级缓冲速度通常更快但需要安装connectorx依赖未安装时会抛出MissingDependencyException该能力同样适用于单表资源sql_table例如结合增量字段加载from dlt.sources.sql_database import sql_table table sql_table( tableorders, backendconnectorx, backend_kwargs{return_type: arrow_stream, batch_size: 100000}, incrementaldlt.sources.incremental(updated_at), )二、Dashboard 新增可视化 Pipeline 运行历史特性概述dlt 的本地 Dashboard 在 1.19 中新增了pipeline 运行历史的可视化视图用于直观展示每次运行的状态、耗时与随时间的变化趋势帮助数据工程师快速判断 pipeline 健康状况、定位失败任务。使用方式在本地运行 pipeline 之后通过 CLI 打开 Dashboard 即可查看运行历史图表dlt pipeline pipeline_name dashboard注意运行历史可视化依赖 dlt 运行期间生成的 trace 与状态数据参见 dlt/pipeline/trace.py 与 dlt/pipeline/state_sync.py因此只有实际执行过pipeline.run()的 pipeline 才会展示出有意义的运行记录。应用价值失败诊断加速从时间轴上快速定位失败运行点再结合日志深入排查运行周期观察查看每次运行的时长波动识别慢查询或资源瓶颈健康度巡检周期性检查运行状态分布及时发现间歇性失败。三、通过 ADBC 将 Parquet 快速导入 MSSQL / MySQL / SQLite特性概述1.19 起dlt 可以使用ADBCArrow Database Connectivity驱动将Parquet 文件直接导入 SQL 数据库MSSQL、MySQL、SQLite。当检测到可用的 ADBC 驱动时Parquet 加载会被自动启用并作为首选方式相比基于INSERT的行级写入可带来10 倍到 100 倍的加载速度提升且比 CSV 回退方案更可靠。自动启用与手动回退由于是自动探测驱动的机制你无需修改任何配置即可在支持的环境中享受加速。如果需要显式退回传统的INSERT加载可以在pipeline.run中指定loader_file_formatinsert_valuespipeline.run( iter(()), # your data generator loader_file_formatinsert_values, )工作机制ADBC 走的是「列式批处理 文件级导入」的通道数据先以 Parquet 文件落盘再由 ADBC 驱动以批量列式方式写入目标库绕开了逐行构造INSERT语句的开销。从加载器实现看文件格式的解析与写入任务由 dlt/load/load.py 与 dlt/destinations/job_client_impl.py 中对应的 SQL 目标实现协同完成ADBC 驱动的可用性决定了自动启用与否。适用前提与限制目标库限定为MSSQL、MySQL、SQLite三类 SQL 数据库环境必须存在对应的 ADBC 驱动例如adbc-driver-sqlite、adbc-driver-postgresql等驱动缺失时 dlt 会回退到原有加载路径若你的业务对写入语句有特殊要求例如完全自定义的 insert 逻辑可显式指定insert_values格式关闭自动加速。四、用Schema.to_mermaid()可视化任意 dlt Schema特性概述1.19 为Schema对象新增了to_mermaid()方法可将任意 dlt Schema 导出为Mermaid ER 图适用于文档编写、Pull Request 评审说明、新成员上手材料等场景。导出结果可直接在 GitHub Markdown、Notion 等支持 Mermaid 渲染的工具中原生展示。核心 APIschema_mermaid pipeline.default_schema.to_mermaid()to_mermaid()定义于 dlt/common/schema/schema.py其内部将 Schema 序列化后委托给 dlt/helpers/mermaid.py 中的schema_to_mermaid生成 Mermaid 字符串。方法签名如下def to_mermaid( self, remove_processing_hints: bool False, hide_columns: bool False, hide_descriptions: bool False, include_dlt_tables: bool True, ) - str:参数默认值作用remove_processing_hintsFalse移除数据处理类 hints如x-normalizer、x-loader标记与冗余信息缩小 Schema 体积、提升可读性hide_columnsFalse隐藏列细节适合表数量多、图幅过大的场景hide_descriptionsFalse隐藏列描述文本include_dlt_tablesTrue是否包含内部 dlt 表_dlt_version、_dlt_loads、_dlt_pipeline_state完整示例import dlt pipeline dlt.pipeline(pipeline_nameevents, destinationduckdb) dlt.resource def events(): yield {event_id: 1, country: DE, ts: 2024-01-01T00:00:00Z} pipeline.run(events()) # 导出为 Mermaid 字符串 print(pipeline.default_schema.to_mermaid()) # 精简模式去掉处理 hints、隐藏描述与内部表 mermaid_clean pipeline.default_schema.to_mermaid( remove_processing_hintsTrue, hide_descriptionsTrue, include_dlt_tablesFalse, )命令行导出除 Python API 外也可以在 CLI 中导出 Schemadlt的 CLI 工具链内部同样使用了to_mermaid参见 dlt/_workspace/cli/utils.py适合在 CI 或文档构建流程中自动化生成 Schema 图。五、Snowflake 聚类键Clustering Key增强特性概述Snowflake 目标在 1.19 中开始支持基于列 hints 更新聚类键。聚类变更会在触发表结构变更时例如新增列自动应用从而无需重建大表即可持续调优聚类策略对超大表的查询性能维护非常友好。使用方法通过apply_hints给列打上cluster: True标记即可dlt.resource(table_nameevents) def events(): yield {event_id: 1, country: DE} events.apply_hints(columns[{name: event_id, cluster: True}]) pipeline.run(events())底层实现逻辑聚类 SQL 的生成位于 dlt/destinations/impl/snowflake/snowflake.py_get_cluster_sql根据列名列表生成CLUSTER BY (col1, col2, ...)语句列名会经escape_column_name转义_get_alter_cluster_sql生成独立的ALTER TABLE ... CLUSTER BY (...)语句_add_cluster_sql区分两种场景建表阶段将CLUSTER BY追加进CREATE TABLE语句而表已存在、需要变更时generate_alterTrue则签发单独的ALTER TABLE语句执行聚类键更新避免整表重建。由此可以确认聚类键的更新发生在 dlt 判定需要变更表结构如新增列触发ALTER时自动随行执行不需要手动重建表。使用要点一个表可以有多个聚类列全部通过cluster: True标记即可聚类键变更属于结构变更操作会随下一次表结构变更触发因此如果只是数据量增长而不涉及结构变化需结合 dlt 的表结构演进机制例如新增字段来触发该能力面向大型事实表、事件表的查询性能优化具体收益取决于查询模式与数据分布。六、其他更新与社区贡献感谢新贡献者1.19 版本收到了来自社区的积极贡献包括 zjacom、JayJai04、anair123、martinibach、hello-world-bfree、tahamuzammil100、timH6502、wrussell1999、luqmansen 等开发者对应 PR 编号 #3294 至 #3332涉及 bug 修复与功能改进。完整的变更列表完整的 1.19 变更清单可参考项目 Release 记录本文聚焦于四项核心用户可见特性特性核心价值相关源码ConnectorX Arrow 流式大结果集增量处理控制内存dlt/sources/sql_database/helpers.pyDashboard 运行历史可视化运行状态与耗时趋势dlt/pipeline/trace.pyADBC 导入 Parquet10–100 倍 SQL 库加载提速dlt/load/load.pySchema.to_mermaid()Schema 一键导出为 Mermaid 图dlt/common/schema/schema.pySnowflake 聚类键更新免重建大表、持续调优聚类dlt/destinations/impl/snowflake/snowflake.py升级建议如果你的sql_database管道经常抽取超大表优先尝试backendconnectorxreturn_typearrow_stream并结合batch_size控制内存水位如果目标库是 MSSQL / MySQL / SQLite 且加载吞吐是瓶颈检查环境是否可安装对应 ADBC 驱动以自动获得 Parquet 批量加载加速在文档与代码评审中推广to_mermaid()把 Schema 演进可视化纳入团队协作流程对 Snowflake 大表用户通过apply_hints(columns[...{cluster: True}])建立聚类策略减少手动 DDL 维护成本。【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考