Python数据预处理:使用Dask和Numba并行化加速

简介: 本文是针对Python设计一种并行处理数据的解决方案——使用Dask和Numba并行化加速运算速度。案例对比分析了几种不同方法的运算速度,非常直观,可供参考。

如果你善于使Pandas变换数据、创建特征以及清洗数据等,那么你能够轻松地使用Dask和Numba并行加速你的工作。单纯从速度上比较,Dask完胜Python,而Numba打败Dask,那么Numba+Dask基本上是无敌的存在。将数值计算分成Numba sub-function和使用Dask map_partition+apply,而不使用Pandas。对于100万行数据,使用Pandas方法和混合数值计算创建新特征的速度比使用Numba+Dask方法的速度许多倍。

Python:60.9x | Dask:8.4x | Numba:5.8x |Numba+Dask:1x

作为旧金山大学的一名数据科学硕士,会经常跟数据打交道。使用Apply函数是我用来创建新特征或清理数据的众多技巧之一。现在,我只是一名数据科学家,而不是计算机科学方面的专家,但我是一个喜欢捣鼓并使得代码运行更快的程序员。现在,我将会分享我在并行应用上的经验。

大多Python爱好者可能了解Python实现的全局解释器锁(GIL),GIL占用计算机中所有的CPU性能。更糟糕的是,我们主要的数据处理包,比如Pandas,很少能实现并行处理代码。

Apply函数vs Multiprocessing.map


%time df.some_col.apply(lambda x : clean_transform_kthx(x))
Wall time: HAH! RIP BUDDY
# WHY YOU NO RUN IN PARALLEL!?

Tidyverse已经为处理数据做了一些美好的事情,Plyr是我最喜爱的数据包之一,它允许R语言使用者轻松地并行化他们的数据应用。Hadley Wickham说过:

plyr是一套处理一组问题的工具:需要把一个大的数据结构分解成一些均匀的数据块,之后对每一数据块应用一个函数,最后将所有结果组合在一起。”

对于Python而言,我希望有类似于plyr这样的数据包可供使用。然而,目前这样的数据包还不存在,但我可以使用并行数据包构成一个简单的解决方案

Dask

之前在Spark上花费了一些时间,因此当我开始使用Dask时,还是比较容易地掌握其重点内容。Dask被设计成能够在多核CPU上并行处理任务,此外借鉴了许多Pandas的语法规则。

现在开始本文所举例子对于最近的数据挑战而言,我试图获取一个外部数据源(包含许多地理编码点),并将其与要分析的一大堆街区相匹配。在计算欧几里得距离的同时,使用最大启发式将最大值分配给一个街区。

最初的apply:


my_df.apply(lambda x: nearest_street(x.lat,x.lon),axis=1)

Dask apply:


dd.from_pandas(my_df,npartitions=nCores).\
   map_partitions(\
     lambda df : df.apply(\
         lambda x : nearest_street(x.lat,x.lon),axis=1)).\
     compute(get=get)
# imports at the end

二者看起来很相似,apply核心语句map_partitions,最后有一个compute()语句。此外,不得不npartitions初始化。 分区的工作原理就是将Pandas数据帧划分成块,对于我的电脑而言,配置是6核-12线程,我只需告诉它使用的是12分区,Dask就会完成的工作。

接下来,将map_partitions的lambda函数应用于每个分区。由于许多数据处理代码都是独立地运行,所以不必过多地担心这些操作的顺序问题。最后,compute()函数告诉Dask来处理剩余的事情,并把最终计算结果反馈给我。在这里,compute()调用Dask将apply适用于每个分区,并使其并行处理。

由于我通过迭代行来生成一个新队列(特征),而Dask apply只在列上起作用,因此我没有使用Dask apply,以下是Dask程序:


from dask import dataframe as dd
from dask.multiprocessing import get
from multiprocessing import cpu_count
nCores = cpu_count()

NumbaNumpyBroadcasting

由于我是根据一些简单的线性运算(基本上是勾股定理)对数据进行分类,所以认为使用类似下面的Python代码会运行得更快一些。


for i in intersections:
    l3 = np.sqrt( (i[0] - [1])**2 + (i[2] - i[3])**2 )
# ... Some more of these
    dist = l1 + l2
    if dist < (l3 * 1.2):
        matches.append(dist)
# ... More stuff
### you get the idea, there's a for-loop checking to see if
### my points are close to my streets and then returning
closest
### I even used numpy, that means fast right?

Broadcasting用以描述Numpy中对两个形状不同的矩阵进行数学计算的处理机制。假设我有一个数组,我会通过迭代并逐个变换每个单元格来改变它

# over one array
for cell in array:
     cell * CONSTANT - CONSTANT2
# over two arrays
for i in range(len(array)):
     array[i] = array[i] + array2[i]

 

相反,我完全可以跳过for循环,并对整个数组执行操作。Numpy与broadcasting混合使用,用来执行元素智能乘积(对位相乘)


# over one array
(array * CONSTANT) - CONSTANT2
# over two arrays of same length
# different lengths follow broadcasting rules
array = array - array2

Broadcasting可以实现更多的功能,现在看看骨架代码:


from numba import jit
@jit # numba magic
def some_func()
     l3_arr = np.sqrt( (intersections[:,0] -
intersections[:,1])**2 +\
                               (intersections[:,2] -
intersections[:,3])**2 )
# now l3 is an array containing all of my block lengths
# likewise, l1 and l2 are now equal sized arrays
# containing distance of point to all intersections
      dist = l1_arr + l2_arr
      match_arr = dist < (l3_arr * 1.2)
# so instead of iterating, I just immediately compare all
of my
# point-to-street distances at once and have a handy
# boolean index

从本质上讲,代码的功能是改变数组。好的一方面是运行很快,甚至能和Dask并行处理速度比较。其次,如果使用的是最基本的Numpy和Python,那么就可以及时编译任何函数。坏的一面在于它只适合Numpy和简单Python语法。我不得不把所有的数值计算从我的函数转换成子函数,但计算速度增加得非常快。

将其一起使用

简单地使用map_partition()就可以将Numba函数与Dask结合在一起,如果并行操作和broadcasting能够密切合作以加快运行速度,那么对于大数据集而言,将会看到运行速度得到大幅提升。


上面的第一张图表明,没有broadcasting的线性计算其表现不佳,并行处理和Dask对速度提升也有效果。此外,可以明显地发现,Dask和Numba组合的性能优于其它方法。

上面的第二张图稍微有些复杂,其横坐标是对行数取对数。从第二张图可以发现,对于1k到10k这样小的数据集,单独使用Numba的性能要比联合使用Numba+Dask的性能更好,尽管在大数据集上Numba+Dask的性能非常好。

优化

为了能够使用Numba编译JIT,我重写了函数以更好地利用broadcasting。之后,重新运行这些函数后发现,平均而言,对于相同的代码,JIT的执行速度大约快了24%。

可以肯定的说,一定有进一步的优化方法使得执行速度更快,但目前没有发现。Dask是一个非常友好的工具,本文使用Dask+Numba实现的最好成果是提升运行速度60倍。如果你知道其它的提升执行速度的技巧,欢迎在留言区分享。


数十款阿里云产品限时折扣中,赶紧点击领劵开始云上实践吧!

作者信息

Ernest Kim,旧金山大学硕士生,专注于机器学习、数据科学。

个人主页:https://www.linkedin.com/in/ernestksocial/

本文由北邮@爱可可-爱生活老师推荐,阿里云云栖社区组织翻译。

文章原标题《Data Pre-Processing in Python: How I learned to love parallelized applies with Dask and Numba》,作者:Ernest Kim,译者:海棠,审阅:袁虎。

文章为简译,更为详细的内容,请查看附件

相关文章
|
4月前
|
数据采集 机器学习/深度学习 数据挖掘
python数据分析——数据预处理
数据预处理是数据分析过程中不可或缺的一环,它的目的是为了使原始数据更加规整、清晰,以便于后续的数据分析和建模工作。在Python数据分析中,数据预处理通常包括数据清洗、数据转换和数据特征工程等步骤。
127 0
|
3月前
|
数据采集 自然语言处理 搜索推荐
python【包含数据预处理】基于词频生成词云图
背景目的 有一篇中文文章,或者一本小说。想要根据词频来生成词云图。
|
2月前
|
机器学习/深度学习 人工智能 数据挖掘
Numba是一个Python库,用于对Python代码进行即时(JIT)编译,以便在硬件上高效执行。
Numba是一个Python库,用于对Python代码进行即时(JIT)编译,以便在硬件上高效执行。
|
3月前
|
数据采集 自然语言处理 搜索推荐
python 【包含数据预处理】基于词频生成词云图
这段文本是关于如何使用Python基于词频生成词云图的教程。内容包括:1) 中文分词的必要性,因中文无明显单词边界及语言单位特性;2) 文本预处理步骤,如移除特殊符号、网址、日期等;3) 使用`data_process`函数清除无用字符;4) `getText`函数读取并处理文本为句子数组;5) 使用jieba分词库进行分词和词频统计;6) 示例代码展示了从分词到生成词云的完整流程,最后展示生成的词云图。整个过程旨在从中文文本中提取关键词并可视化。
72 5
python 【包含数据预处理】基于词频生成词云图
|
2月前
|
数据采集 机器学习/深度学习 数据可视化
关于Python数据分析项目的简要概述:从CSV加载数据,执行数据预处理,进行数据探索,选择线性回归模型进行训练,评估模型性能并优化,最后结果解释与可视化。
【7月更文挑战第5天】这是一个关于Python数据分析项目的简要概述:从CSV加载数据,执行数据预处理(填充缺失值,处理异常值),进行数据探索(可视化和统计分析),选择线性回归模型进行训练,评估模型性能并优化,最后结果解释与可视化。此案例展示了数据科学的典型流程。
52 2
|
3月前
|
数据采集 机器学习/深度学习 数据挖掘
使用Python进行数据预处理与清洗的最佳实践
本文探讨了Python在数据预处理和清洗中的关键作用。预处理包括数据收集、整合、探索、转换和标准化,而清洗则涉及缺失值、重复值、异常值的处理及数据格式转换。文中提供了使用pandas库进行数据读取、缺失值(如用平均值填充)和重复值处理、异常值检测(如IQR法则)以及数据转换(如min-max缩放)的代码示例。此外,还讲解了文本数据清洗的基本步骤,包括去除标点、转换为小写和停用词移除。整体上,文章旨在帮助读者掌握数据预处理和清洗的最佳实践,以提高数据分析的准确性和效率。
144 2
|
4月前
|
机器学习/深度学习 数据采集 算法
Python中的机器学习入门:从数据预处理到模型评估
Python中的机器学习入门:从数据预处理到模型评估
280 35
|
4月前
|
机器学习/深度学习 数据采集 PyTorch
在Python中进行深度学习的数据预处理
在Python中进行深度学习的数据预处理
54 2
|
数据采集 Python 数据可视化
[Python] 数据预处理(缺失值、异常值、重复值) [相关方法参数说明、代码示例、相关概念](三)
[Python] 数据预处理(缺失值、异常值、重复值) [相关方法参数说明、代码示例、相关概念](三)
|
数据采集 Python
[Python] 数据预处理(缺失值、异常值、重复值) [相关方法参数说明、代码示例、相关概念](二)
[Python] 数据预处理(缺失值、异常值、重复值) [相关方法参数说明、代码示例、相关概念](二)
下一篇
云函数