CDA数据分析师 出品相信很多人在处理算法的时候都会被那种因为数据量太庞大, 导致计算任务多得让人头大的痛苦经历所折磨, 这种情况真的挺让人发愁的。接下来我们主要围绕四个方法来帮助大家把计算运行的时间给缩短一些, 好让大家在调试算法少花些等待的时间。今天我们来说明第四方面也就是最后这一方面的内容, 主要就是讲述关于应用Dask这个工具的具体方法。1.简介现在, 人们对机器学习算法并行化的需求变得越来越多了。因为数据的大小甚至模型的大小都在呈指数级增长。如果我们有一个工具, 这个工具可以帮助我们去处理并行化的情况, 它不仅可以并行化处理Numpy的计算, 而且还可以并行化处理我们的机器学习算法, 这些算法可能是来自其他来源的算法, 这样我们就不会遇到太多麻烦, 这个工具对我们来说会非常有帮助。好消息是确实存在这样的库, 其名称为Dask。Dask是一个并行计算库。它不仅有助于并行化现有的机器学习工具, 也有助于并行化低级任务或者功能。它还可以使用高级集合来进行并行化处理。此外它可以通过制作任务图来处理这些功能之间存在的复杂交互。也就是说, 使用的是低级调度程序。这属于这种类型的线程或多处理模块。他们同样拥有一个名称叫做dask-ml的独立的机器学习库, 该库可以与其他的现有库进行集成, 比如那些已经被提及的库和另外一些未指定名称的库。Dask会把分配给它的那些任务, 通过绘制任务之间交互的那个图的方式来进行并行化处理。使用Dask的()方法来查看一下你自己的工作状况, 将会是非常有帮助的一件事情, 因为这个方法是可以用于各种各样的数据类型之上的, 同时也是可以用于你计算复杂任务链的场合之下的。这个方法会把这个任务图给你输出出来, 假如你的每个层级里面都有好几个节点的话, 那么意思就是你的那个任务链的结构在很多个层次上都存在不少相互独立的任务的情况, 比如像是在数据处理块上面进行并行地处理一些任务的场景这样去理解, 然后这样的话, Dask这个工具就可以让这些部分同时运行起来以达到并行化处理的目的了。请注意, Dask这个项目还是相对较新的, 它还有很长一段路要走。但是呢, 如果你并不想要去学习一个完全崭新的API例如, 那么Dask就是你的最佳的选择, 它在将来肯定还会变得越来越好的, 不过从当前的情况来看, Spark/仍然是遥遥领先的态势, 并且也依然会继续去进行改进工作。这是一个非常完善的项目。2.数据类型在Dask这个工具里, 每一种数据类型都有自己的分布式版本, 这些版本分别对应着里的list、以及numpy库里的数组。你要注意, 这些数据的大小有可能超过你的电脑内存限制, 而Dask会选择一种叫作并行计算的方法来运行这些任务。你可以这样去理解它的工作模式, 也就是通过执行很多次比较小的计算任务来最终完成那个大型的计算过程, 这些小任务是以数据块为基本单位展开的, 至于具体要分成多少个这样的数据块, 那就要看总的数据是怎么分割的了。a数组在网格结构里面, 包含有大量的Numpy数组, 而这些数据整体被用作了Dask数组。Dask Array处理的是体量极为庞大的人工数据, 它把这些巨大的数组切割成多个小块, 并且安排让每一个块都在不同的线程里同步展开计算。这个功能库里已经整合了不少现有的numpy标准操作指令, 使用者可以调动这些指令来达到提速的目的, 然而需要警惕的是, 那一大堆功能之中确实存在某些方法尚未得到具体的编码实现。只要支持numpy切片, Dask Array就可以从任何类似数组结构中读取数据, 并且通过使用, 以及通过使用Dask . Array .方法它可以具有.shape属性, 它还可以读取.npy和.zarr文件。dask.array as numpy as nparr numpy..(1, 1000, (10000, 10000))darr da.(arr, (1000, 1000))# 它会生成大小为(1000,1000)的块darr.# 100当你的数组真的是非常沉重的时候, 换句话说, 就是这些数组根本无法被放入内存当中去, 而在这种情况下, NumPy这个工具也是根本无能为力的, 所以是可以使用Dask来进行处理的。因此呢, Dask就把它们分成了很多的数组块, 并且还会替你并行地进行处理操作。目前, Dask对每个方法都采用了惰性计算的模式。所以, 假如想要真的算出函数的结果, 就必须得使用那个点后面跟着括号的调用方法啦。它会按照小块的方式把结果给并行地算出来, 并且同时也对每一个单独的进行并行的处理操作。 pute()1当元素数量较少的时候, Numpy比Dask更快地执行处理任务其次, Dask接管了原本由Numpy负责的工作规模耗时大约是1e7个元素的情况最后Numpy无法产生更多元素的处理结果原因是它无法将这些数据存储在内存里面。b提供5份数据, 这些数据和单个Dask实例有关联, 它们展示的是每个月的状况数据来源可以是不同的diff文件Dask的做法和Dask有点像, 就是把大文件切成小块, 然后对这些块的计算函数并行地处理, 这样就能对太大的数据文件进行并行计算了。dask. as dddf dd.(“(s).csv”, 50e6)现在, 你可以去应用库里面大多数的功能, 把这些功能都使用上, 然后将它们在此处进行应用的这种操作。agg df.(“”).(计算它们的总和, 以及平均值、最大值和最小值。先把那些计算的值给汇总起来, 然后再去查一下那张大表格是怎么把它们给合并在一起的, 合并的时候用的是左边那种方式, 最后再拿出来前面几行看看。cBagDask这个框架里面的Bag包能够并行处理那种包含了好几种不同类型数据的列表相似的对象。当你去尝试处理一些半结构化的数据的时候, 这个功能会非常有用, 比如说JSON格式的Blob或者是日志文件。引入dask中的bag工具并将其命名为 dbb 变量, 然后通过该方法去读取名为 a.txt 的文件对象, 紧接着使用.take方法并传入参数 1, 以此获取第一行的数据。把数据一行一行地读进去之后, 用.take方法把指定数量的行提取出来并且以元组的样式给打印出来。Dask Bag 在这种类型的对象集合上面, 实现了像 map, , fold, 还有这些操作之类的功能。它会利用迭代器, 并行地把这些事情给做完, 而且所占用的那种内存是非常小的。它就相当于某个东西的并行版本, 或者是 RDD 的那个版本的相似存在一样。 b.( x: x“Name”“James”)\ .map( x: x“” “”)pute()3.延时如果你的任务情况属于比较简单的类别, 并且你处于能力不足的状态或者主观上不愿意去运用那些被称为高级集合的工具来进行相关操作的话, 那么你是可以使用那些被称为低级调度程序的东西的, 这个低级的东西能够给你提供帮助, 帮助你通过使用接口名称为dask的这种手段来实现对你代码以及你的算法进行并行化处理的操作, 此外, 由名字拼写为dask的这个对象也是具备执行延迟计算这一功能的。Dask中的delay函数可以这么理解: 它能让普通的代码延迟执行, 变成一个未来要计算的图任务节点。比如定义sq这个函数的平方操作时, 先写x*x再调用delay装饰器。再比如加法的add函数把参数相加后也用了delay处理。最后求和的sum函数遍历列表逐个累加元素。sum你可以依据需求在涉及这些函数的操作中设置复杂的关联互动, 并且能够把前一个任务所产出的结果, 用来作为后续任务的输入参数, Dask平台并不会即刻执行对这些函数的计算操作, 而是会把你们所要完成的任务构建为一张图表呈现出来, 通过这种方式有效地将你所调用函数之间的关联性进行整合处理。将外接程序dask加入到列表中, 具体的做法是让一个列表等于从1到10这个范围内生成的数值集合。for i in range(len()): temp.(sq(先计算输入的sq然后保存, 接下来执行延迟计算操作在列表这个变量上, 将temp这个变量初始化为临时值最后再把结果赋值给temp。循环变量从零开始, 每次递增二直到小于列表长度减一这一范围, 在每一次循环迭代的过程中调用临时对象的添加方法。i1添加两个连续的值, 计算结果是从前面一个步骤延续下来的, 然后等于求和, 把之前所有步骤的汇总结果加起来进行计算。你可以在任何可以进行并行处理并且拥有众多小型任务的代码模块中引入延迟机制, 从而实现执行效率的提升目标。这些待处理的计算任务不仅涵盖多种你计划实施的运算情形, 就像前述所提及的典型实例一样, 还可以涉及通过同步并发方式对海量数据文件进行批量读取等操作场景。4.分布式起初, 我们在计算任务所得结果时, 一直都在使用Dask提供的默认那种调度方式。不过, 若你想要换一种用法, 是可以根据实际具体需要, 去把这种设置改成Dask里头所给出的其他备选项目的。Dask 带有四个可用的调度程序这个调度程序是由线程池来支持的。该调度程序是由进程池来进行支持的。它是被称为同步调度程序的东西, 这个名称也叫做sync, 它的主要用途是用来进行调试工作。它是用来在多台计算机上进行图形分布式调度的程序。pute(也就是减号), 这个操作可以用来调试, 或者可以使用dask来设置并调用函数, 这里需要注意一下, 根据来自官方网页上的说法, 如果那个被称作全局解释器锁, 也就是简称为GIL的函数释放掉了的话, 那么线程任务就会运行得非常顺畅, 但是多线程处理的情况总是会存在启动时间比较慢这个问题, 并且在各种任务之间还需要耗费大量的通信资源。#你可以通过其中一个得到调度程序, 例如 :dask..get, 又或者是 dask.local 这个最后的那个选项, 它是属于单线程类型的。但是, Dask这个东西里面, 居然还另外配备了一个叫做dask的调度器, 这个事儿, 你得知道。就因为下面说到的那些原因, 所以它反而有可能是大家最倾向于去使用的那一种方案。1. 它在访问异步API的时候, 特别提供了这种访问的能力。1. 它提供了一个诊断仪表板, 这个仪表板可以提供有价值的见解, 这种见解是关于性能方面的, 同时也是关于进度状况的。1. 它在处理数据位置方面, 具备应对更为复杂情形的能力, 所以说, 针对那种需要多个流程来处理的工作负载场景来说, 其实际表现要比多进程调度机制来得更加高效。你可以在导入相关内容之后接着创建一个叫做 Dask 的调度器, 具体的实现过程是建立一个客户端。从dask库中调用的内容等于空括号, 这是为了设置本地环境, 当用户拥有Bokeh软件包并且正确安装之后, 可以前往特定的网页地址进行查看, 这个网页地址就是8787前面的部分加上斜杠以及dash开头的字样, 目的是让用户能够访问诊断类的仪表板以获取相关信息。现在, 你可以调用方法, 把函数和参数当作该方法的输入值传进去, 以此来把具体的任务提交到这一个集群里面去。接下来, 我们还可以使用另外的某个方法, 或者再使用一个其他的方法, 把计算之后得到的结果收集起来。设置一个变量叫做sent, 让它等于sq乘以4的积, 这里的sq是一个函数, 或者也可以直接用点连接的方式来调用这个函数的方法, 把结果赋给sent。另外, 你完全可以直接采用 dask.. 这种方式, 来实时查看位于当前单元格里面的那个任务, 它究竟进行到了怎样的进度环节。同理, 你还能够明确地做出选择, 去调用 dask..wait 这一功能, 以此来耐心等待并完成手头那些任务的处理过程。Note: (Local )有时您会注意到Dask正在超出内存使用即使它正在划分任务。它可能发生在您身上因为您试图在数据集上使用的函数需要您的大部分数据进行处理而多重处理可能使情况变得更糟因为所有工作人员都可能试图将数据集复制到内存中。这可能发生在聚合的情况下。也许您心里会有这样的想法, 那就是去限制一下Dask这个工具, 让它仅仅只能够使用特定数目的内存。在这些状况下面, 您能够通过运用Dask当中的那些参数, 并且把相关的具体数值提交到相应的函数括号里面去, 以此来实现基于本地点机内部所具备的核心能力进行构建操作。from dask. , (1, 1, False, ‘25GB’, 0, True, 0)参数等于零与另一个空格加上参数等于零, 将会为这一特定的客户端挑出一个随机的端口号。如果是参数等于假的状态, 那么分布式任务系统的客户端就不会复制数据集合, 这件事可能就会在你每一个建立起来的进程中发生。你可以按照你自己心里想要的样子或者受到的一些限制, 去调整那个客户端的表现方法。要是你想明白更多的信息的话, 可以去看看那些参数的情况介绍。同样的道理, 你也可以在同一台计算设备上面, 用不一样的端口号, 同时开启好几个客户端来运行程序。5.机器学习Dask它也存在有一个专门的库, 这个库的作用是去帮助并且去允许那些大多数的非常流行的机器学习库, 例如像说有的这些库能够进行运行和操作。当你在进行机器学习项目的时候, 你是有可能遭遇多个不同层面的问题的存在的。具体应当选用什么样的策略方法, 这主要是要看你所面对的那个具体问题到底是怎样的一个状况的。1. 针对大型模型这种情况, 适合将数据放入到内存里面去操作, 可是需要花费很长的时间来进行训练工作, 这中间涉及到非常多的超参数组合情况, 还包含数量众多的模型大型集合等等因素。1. 这是一个大型数据集, 数据的内容大于随机存取内存的大小, 并且不能选择抽样。因此, 你就需要做到以下几点内容。· 对于内存中适合的问题只需使用-learn或你最喜欢的ML库即可对于大型模型, 请选择使用点号和你所偏好的一系列学习的估计器。在处理规模较为庞大的数据集合时, 建议采取使用估计算法这一做法。a预处理它里面包含了一些功能, 像是稳健标量、标准标量、标签编码器、独热编码、多项式特性这些, 另外也有它特有的一些功能, 比如分类器、虚拟编码、序数编码器这个样子的。你可以像使用一样使用它们。从数据源读取csv文件, 限制前50万行数据, 最终形成数据框对象。“” rsc.(df“”你能够运用Dask平台所提供的数据预处理功能, 借此手段来构建出一个处理管道。b超参数搜索:Dask具备用来执行超参数搜索方面的功能机制, 好比说是某某方法呀或者是那样一类方法之类的这样一些情况内容存在其中。从已有的内容那里, 我们得到了 xtr, ytr 和 xval, yval。其中 y 等于五万乘以 xtr。接下来是 (X, y)。最后是 (,)。使用 cv 等于十。用 .fit(xtr, ytr) 进行拟合。并且, 在打算搭配着用来搞估算器这个东西的时候, 那么去选用dask-ml这个方案也是能够行的通的。注意, 如果是从Dask而来的话。一旦你要去使用后拟合任务。像是一起评分或者把预测结果给搞出来这些操作。那么你就得去使用那个基础估计量的评分方法才行。再者说呢, 如果你的估算器本身它可能来自某些地方。然后它还没办法去处理那些大型的数据集的话。你就应该把这个估算器放到双引号里面去把它给包装起来。虽然原文中有些地方是空的或者是用双引号指代。但意思就是要这么做。这样做以后它就可以实现并行化了。它可以并行化处理诸如一些方法的调用。这些方法可以是获取数据之类的动作。也可以是训练模型等等的行为。总之就是让计算过程能够并行执行从而提高效率。c也就是那个模型, 或者说就是那个估计器:Dask拥有线性模型的诸多特性, 例如支持诸如线性回归以及逻辑回归之类的方法, 它还具备聚类模型的能力, 涵盖像K均值聚类以及层次聚类这样的功能, 同时提供某种专门针对数据进行操作的技术手段, 并且存在一套依托于Dask平台来执行模型训练工作的具体流程和方法。如果训练数据较小, 则可以把模型以及Dask这个工具结合在一块使用, 要是包装数据规模比较大, 那么就可以让某个未具体命名的包装器来进行处理, 而要是测试数据规模很大, 则可以考虑把之前的那些方法和新的方法组合在一起进行应用。从。.. ()el.fit(,)preds是 el。(Xtest)。如果面临的情况是数据集规模并不庞大, 但是机器学习模型的参数量却非常巨大, 那么在这种情况下是可以利用这种方法的。人们已经编写了许多专门用于并行计算的算法, 其中有些人可能接触过等于负一这个特定的参数值, 这些算法借助了多线程和多进程技术, 从而实现了工作负载的并行化处理。如果要使用Dask来完成类似的并行化任务, 那么第一步必须是创建一个被称为客户端的对象, 然后需要使用方法将相关的核心代码块包裹起来, 并且在这个方法的作用期间要使用字符串“dask”作为具体的配置项。用带有dask的learn代码, 让程序在后台慢慢跑, 别卡死。请注意, DASK 后端对扩展 CPU绑定的工作负载而言是非常有用的。这适用于那些需要在 RAM中存放数据集, 但同时拥有大量能够并行完成的操作的工作负载。如果要扩展到受到限制于 RAM的工作负载, 比如数据超过内存规模的情况, 你就应当采用 Dask所内置的模型和方法。另外, 假如你手里用来训练的数据量特别巨大, 大到压根儿就装不进电脑的内存里头的情况之下, 那你就得动用Dask这个工具里面自带的那些估算器了, 这样一来速度能明显提升上去。还有事儿呢, 你也可以去用Dask提供的其他功能, 它这一招其实就是借用了那些底层基础估算器所使用的那个处理方法, 对整体性的数据集合展开训练工作, 不过这个训练的过程在实际操作当中是连续不断进行的。Dask的内置估计器能够很好地针对大型数据集进行扩展, 它还支持与包括admm、lbfgs在内的多种优化算法配合使用, 同时也兼容如L1、L2之类的正则化器处理。这段文字的意思是调用lr对象的fit方法, 将变量X和变量y作为参数传入。经历了四个阶段的内容解说, 你现在应该已经懂得该如何去提升算法的运转速度了呢。在疫情发生的当下, 以前那些急急忙忙的脚步, 最后终于稍微地停顿了一下, 这也正好是一个合适的时刻, 让我们能够非常认真地、反复地思考一下关于我们自己未来的职业打算是什么, 以及关于整个人生的大方向应该如何去规划。如果我们能够比其他人更早地提前做好各种准备, 就像是提前把雨伞带在身边以防下雨一样, 为以后的日子积攒下足够的能量储备, 那就是为了将来的爆发做好了一切准备