数据分析数据工程大数据【免费下载链接】modinModin: Scale your Pandas workflows by changing a single line of code项目地址https://gitcode.com/gh_mirrors/mo/modin点击查看免费下载Modin 通过统一的 API 层支持多种分布式执行引擎其中 pandas on Dask 是让 pandas 工作负载跑在 Dask 调度器与 worker 之上的关键执行后端。本文以仓库文档 docs/flow/modin/core/execution/dask/implementations/pandas_on_dask/index.rst 为主线结合源码剖析 Data Transformation数据变换、Data Ingress数据读入与 Data Egress数据写出三条通路的完整调用链以及PandasOnDaskDataframe、PandasOnDaskDataframePartition、虚拟分区与分区管理器四类核心组件的职责与实现细节。读完本文你将能够理解一条用户 DataFrame API 调用在 Dask 引擎下从查询形成、算子映射、远程任务下发到结果物化的全过程并掌握如何在 Modin 中启用与排查 pandas on Dask 执行。启用 pandas on Dask 执行Modin 默认以 Ray 为执行引擎切换为 pandas on Dask 只需在导入modin之前设置环境变量或在脚本中显式指定export MODIN_ENGINEdask python your_script.py也可以在使用pandas前通过配置 API 切换import modin.config as cfg cfg.Engine.put(dask) # 或在环境变量中设置 MODIN_ENGINEdask import modin.pandas as pd完整的安装与启动指引含本地与集群模式、Dask 调度器/worker 的部署方式参见 docs/development/using_pandas_on_dask.rst 的 usage 章节——这也是 index.rst 明确给出的启用入口。切换之后所有执行 pandas on Dask 语义的数据变换、读入与写出查询都会按下文描述的组件链路流过。数据变换Data Transformation一条查询的完整生命周期上图展示了 pandas on Dask 执行下数据变换查询从 API 层到执行层的组件通路来自文档原图。当用户调用任意modin.pandas.dataframe.DataFrameAPI 时一条查询在API 层开始形成并最终在执行层被真正执行API 层modin.pandas.dataframe.DataFrame负责对查询做恰当的预处理——例如判断最终结果是DataFrame还是Series对象同时负责净化sanitize交给查询编译器的输入例如校验查询中的某个参数、定义特定的中间值以便给查询编译器提供更多上下文。PandasQueryCompilermodin/core/storage_formats/pandas/query_compiler.py接收来自 API 层的查询决定如何将其作用到数据的一个子集上——按单元格cell-wise处理或按轴axis-wise作用于以 pandas 存储格式为后端的某个分区。它将查询映射到 Core Algebra Operators 中的某一个代数算子。PandasOnDaskDataframemodin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py作为 Dask 引擎下 dataframe 代数的具体实现它从PandasDataframemodin/core/dataframe/pandas/dataframe/dataframe.py继承通用功能是查询编译器与PandasOnDaskDataframePartitionManager之间的中间层。PandasOnDaskDataframe 的源码级细节从源码看PandasOnDaskDataframe的实现非常精简核心职责是绑定引擎专属的分区管理器与标识引擎身份class PandasOnDaskDataframe(PandasDataframe): _partition_mgr_cls PandasOnDaskDataframePartitionManager property def engine(self) - str: return Dask_partition_mgr_cls指向PandasOnDaskDataframePartitionManager即所有分区分块/重组/函数下发都经由 Dask 分区管理器完成。engine属性返回Dask供上层如日志、指标与报错信息识别当前执行引擎。该类还通过自定义的__reduce__/reconnect方法解决了一个 Dask 特有的分布式问题当对象随任务序列化到 worker 进程后reconnect会利用调度器地址default_client().scheduler_info()[address]在 worker 进程中重建default_client保证后续远程任务仍能正确路由到调度器。PandasOnDask 的四类核心组件Modin 为 pandas on Dask 执行专门实现了Dataframe、PartitionManager、AxisPartition与Partition四类组件它们之间的包含关系为PandasOnDaskDataframedataframe.rstPandasOnDaskDataframePartitionpartitioning/partition.rstPandasOnDaskDataframeVirtualPartitionpartitioning/virtual_partition.rstPandasOnDaskDataframePartitionManagerpartitioning/partition_manager.rst块分区 PandasOnDaskDataframePartitionPandasOnDaskDataframePartitionmodin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py是PandasDataframePartition的 Dask 实现提供在单个块分区即一个pandas.DataFrame上执行操作的能力。除了包装一个pandas.DataFrame外该类还持有三项元数据元数据含义length被包装的pandas.DataFrame的行数width被包装的pandas.DataFrame的列数ip持有该pandas.DataFrame的节点 IP 地址注意这里被包装的 DataFrame实际上是一个distributed.Future源码构造函数的data参数断言isinstance(data, Future)也就是说分区对象持有的是分布式内存中的引用数据本身尚未物化到本地。块分区上的操作有两种执行模式异步asynchronously通过apply方法partition.py#L67立即把函数作为远程任务下发。惰性lazily通过add_to_apply_calls把函数追加到call_queue等待后续统一批量下发执行。apply的实现体现了 Dask 后端的性能优化思路它维护一个call_queue当队列中累积了多个函数时通过apply_list_of_funcspartition.py#L351在一次远程任务里串行执行整条调用队列避免多次网络往返而队列长度为 1 时则走更快的apply_func单函数路径partition.py#L320。两者都会通过DaskWrapper.deploy下发到 worker并返回(结果, worker 节点 IP)二元组IP 由distributed.utils.get_ip()在 worker 进程内取得。其他值得关注的方法drain_call_queue把累积的调用队列一次性执行并把结果写回self._data是触发实际计算的关键点。wait先drain_call_queue再阻塞等待远程 future 完成不物化数据。mask惰性构造行/列标签掩码且对 slice 形式的标签提供了快速路径全轴取用直接复用缓存的 length/width否则用compute_sliced_len计算切片后的长度。put把本地对象散列scatter进分布式内存并包装成分区同时记录其length与width。preprocess_func用DaskWrapper.put(func, hashFalse, broadcastTrue)把函数广播到所有 worker使函数对象可被远程执行。length/width/ip三个访问器都支持materializeTrue/False——为False时可能返回尚未物化的distributed.Future为True时强制物化为 int/str。虚拟分区 PandasOnDaskDataframeVirtualPartitionPandasOnDaskDataframeVirtualPartitionpartitioning/virtual_partition.py是PandasDataframePartition在轴列轴或行轴上的实现轴分区是存储在该类中的一个块分区列表的包装器。它派生出两个具体子类PandasOnDaskDataframeColumnPartition按列组织的分区用于需要整列参与的计算如按列的 reduce、groupby 的部分阶段。PandasOnDaskDataframeRowPartition按行组织的分区用于需要整行参与的计算。这种块分区 虚拟轴分区的双层设计使 Modin 在数据变换时能够灵活切换按块cell-wise与按轴axis-wise的执行粒度。分区管理器 PandasOnDaskDataframePartitionManagerPandasOnDaskDataframePartitionManagerpartitioning/partition_manager.py是PandasDataframePartitionManager的 Dask 实现负责分区操控以及向块/行/列分区应用函数。源码中它通过类属性绑定各组件class PandasOnDaskDataframePartitionManager(PandasDataframePartitionManager): _partition_class PandasOnDaskDataframePartition _column_partitions_class PandasOnDaskDataframeColumnPartition _row_partition_class PandasOnDaskDataframeRowPartition _execution_wrapper DaskWrapper它覆写了wait_partitions把二维分区矩阵中的全部底层 futurepartition.list_of_blocks收集起来交给DaskWrapper.wait并行阻塞等待用于如compute、to_pandas前强制所有远程任务完成的场景。数据读入Data Ingress并行读取文件的调用链上图展示了 pandas on Dask 执行下IO 函数从 API 层经工厂分发到远程解析的组件通路来自文档原图。当用户调用modin.pandas.io模块中的任意 IO 函数时API 层查询FactoryDispatchermodin/core/execution/dispatching/factories/dispatcher.py由其根据当前引擎分发到PandasOnDaskFactorymodin/core/execution/dispatching/factories/factories.py。工厂暴露PandasOnDaskIO类modin/core/execution/dask/implementations/pandas_on_dask/io/io.py其职责是从/向文件执行并行的读与写。PandasOnDaskIO通过__make_read/__make_write工厂方法动态组合Dispatcher Parser DaskWrapper来批量生成读写方法实际支持的接口包括读read_csv、read_fwf、read_json、read_parquet、read_feather、read_sql、read_excel写to_parquet、to_sql实验性接口pandas 原生没有的read_csv_glob、read_parquet_glob、to_parquet_glob、read_json_glob、to_json_glob、read_xml_glob、read_pickle_glob、read_custom_text、read_sql_distributed以读取 CSV 为例PandasOnDaskIO.read_csv会把用户查询转发到CSVDispatcher._readmodin/core/io/text/csv_dispatcher.py该处完成两件准备工作参数预处理检查查询参数是否被当前执行支持若不支持则回退default到 pandas 单机实现公共元数据计算计算所有待读分区共用的元数据如 dtype 推断结果等。随后文件被切分为行块row chunks并通过DaskWrapper.deploymodin/core/execution/dask/common/engine_wrapper.py在 Dask worker 上启动远程任务每个 worker 上由PandasCSVParsermodin/core/storage_formats/pandas/parsers.py解析各自的数据块。远程任务全部完成后进行结果后处理最后返回携带已读数据的新查询编译器PandasQueryCompiler供上层 DataFrame 使用。PandasOnDaskIO还提供三类跨引擎互操作方法from_dask(dask_obj)把dask.dataframe.DataFrame转成 Modin 查询编译器先to_delayed()再client.compute得到 future 列表最后按行组装分区to_dask(modin_obj)反向转换用from_delayed构造 Dask DataFrameSeries 场景会先在 worker 上把单列 DataFrame 转成 Seriesfrom_map(func, iterable)把一个映射函数作用于可迭代对象每个元素生成一个行分区构造查询编译器。数据写出Data Egress并行写出文件的调用链上图展示了 pandas on Dask 执行下数据写出查询的组件通路来自文档原图。以写出 CSV 为例PandasOnDaskIO处理用户查询后在 Dask worker 上执行随后请求PandasOnDaskDataframe把数据分解为行方向的分区row-wise partitions这些分区将在 Dask worker 上被并行写入文件。注意文档原文提示目前 pandas on Dask 执行的数据写出egress使用默认的 pandas 单机实现见 index.rst#L82-L83。这意味着to_csv等写出操作当前并不会以完全分布式的形式并行落盘评估写出性能时应以此为前提。引擎级任务调度核心DaskWrapper三条通路的远程任务最终都汇聚到DaskWrappermodin/core/execution/dask/common/engine_wrapper.py。它是 pandas on Dask 执行与 Dask 集群交互的封装层四个核心方法对应 Dask 的四个原语方法底层调用作用deploy(func, f_args, f_kwargs, return_pandas_df, num_returns, pure)client.submit把函数下发到 worker 进程执行num_returns 1时用额外的submit(lambda tup, i: tup[i], ...)拆分多返回值pureFalse用于函数有副作用的场景materialize(future)client.gather把 future 物化为本地对象put(data)client.scatter把数据散列进分布式内存dict 类型会包一层UserDict以规避序列化 bugwait(obj_ids, num_returns)distributed.wait阻塞等待 future 完成而不物化数据num_returns小于总数时用FIRST_COMPLETED循环取前 N 个完成的结果deploy还处理一个特例当传入的func本身是distributed.Future时会经由_deploy_dask_func包装并可按return_pandas_df把结果强制转为pandas.DataFrame。get_dask_client则保证无论代码运行在调度器客户端还是 worker 进程内都能拿到正确的default_client。小结与源码导航pandas on Dask 执行的完整通路可概括为数据变换DataFrameAPI →PandasQueryCompiler→ Core Algebra 算子 →PandasOnDaskDataframe经PandasOnDaskDataframePartitionManager调度PandasOnDaskDataframePartition/ 虚拟分区→DaskWrapper.deploy→ Dask worker。数据读入IO API →FactoryDispatcher→PandasOnDaskFactory→PandasOnDaskIO→ 各*Dispatcher._read→DaskWrapper.deploy→Pandas*Parser远程解析 → 新PandasQueryCompiler。数据写出PandasOnDaskIO→ 按行分解分区 → 并行写出当前默认回退 pandas 单机实现。建议继续深入阅读的仓库路径文档index.rst、dataframe.rst、partitioning/partition.rst、partitioning/virtual_partition.rst、partitioning/partition_manager.rst源码dataframe.py、partition.py、partition_manager.py、io.py、engine_wrapper.py上层文档使用 pandas on Dask、Core Algebra 算子赞分享数据分析数据工程大数据【免费下载链接】modinModin: Scale your Pandas workflows by changing a single line of code项目地址https://gitcode.com/gh_mirrors/mo/modin点击查看免费下载相关推荐Modin 系统架构深度解析分层设计、查询编译链路与分布式执行原理Modin 系统架构深度解析分层设计、查询编译链路与分布式执行原理 本文基于 docs/development/architecture.rst https:数据分析数据工程大数据Modin Pandas on Ray 执行引擎全解析数据转换、数据读取与数据写出的分布式执行链路Modin Pandas on Ray 执行引擎全解析数据转换、数据读取与数据写出的分布式执行链路 本文围绕 Modin 的 PandasOnRay 执行路径数据分析数据工程大数据Modin 存储格式与查询编译器架构解析从 BaseQueryCompiler 到 pandas 存储格式实现Modin 存储格式与查询编译器架构解析从 BaseQueryCompiler 到 pandas 存储格式实现 导读 存储格式Storage Format数据分析数据工程大数据上一篇3个步骤实现智慧树自动刷课完整zhihuishu插件使用指南下一篇智慧树自动学习插件3大核心功能让网课效率翻倍的终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
