视频1 视频21 视频41 视频61 视频文章1 视频文章21 视频文章41 视频文章61 推荐1 推荐3 推荐5 推荐7 推荐9 推荐11 推荐13 推荐15 推荐17 推荐19 推荐21 推荐23 推荐25 推荐27 推荐29 推荐31 推荐33 推荐35 推荐37 推荐39 推荐41 推荐43 推荐45 推荐47 推荐49 关键词1 关键词101 关键词201 关键词301 关键词401 关键词501 关键词601 关键词701 关键词801 关键词901 关键词1001 关键词1101 关键词1201 关键词1301 关键词1401 关键词1501 关键词1601 关键词1701 关键词1801 关键词1901 视频扩展1 视频扩展6 视频扩展11 视频扩展16 文章1 文章201 文章401 文章601 文章801 文章1001 资讯1 资讯501 资讯1001 资讯1501 标签1 标签501 标签1001 关键词1 关键词501 关键词1001 关键词1501 专题2001
Python+大数据计算平台,PyODPS架构搭建
2020-11-27 14:17:16 责编:小采
文档
数据分析和机器学习

大数据基本都是建立在Hadoop系统的生态上的,其实一个Java的环境。很多人喜欢用Python和R来进行数据分析,但是这往往对应一些小数据的问题,或者本地数据处理的问题。如何将二者进行结合使其具有更大的价值?Hadoop现有的生态系统和现有的Python环境如上图所示。

MaxCompute

MaxCompute是面向离线计算的大数据平台,提供TB/PB级的数据处理,多租户、开箱即用、隔离机制确保安全。MaxCompute上主要分析的工具就是SQL,SQL非常简单、容易上手,属于描述型。Tunnel提供数据上传下载通道,不需要经过SQL引擎的调度。

Pandas

Pandas是基于numpy的数据分析的工具,里面最重要的结构是DataFrame,提供一系列绘图的API,背后是matplotlib的操作,非常容易和Python第三方库交互。

PyODPS架构

PyODPS即利用Python进行大数据分析,其架构如上图所示。底层是基础API,可以利用其操作MaxCompute上的表、函数或者资源。再上面是DataFrame框架,DataFrame包括两部分,一部分是前端,定义了一套表达式的操作,用户写的代码会转化成表达式树,这与普通的语言是一样的。用户可以自定义函数,也可以进行可视化,与第三方库进行交互。后端最下面是Optimizer,其作用是对表达式树进行优化。ODPS和pandas都是通过compiler和analyzer提交到Engine来执行。

背景

为什么要做DataFrame框架?

对于任何一个大数据分析工具,都会面临三个维度上的问题:表达力,API、语法、编程语言是否简单、符合直觉?数据,存储、元数据是否能压缩、有效?引擎,计算的性能是否足够?所以就会面临pandas和SQL两个选择。

如上图所示,pandas的表达力非常好,但是其数据只能放在内存中,引擎是单机的,受限于本机的性能。SQL的表达力有限,但是可以用于大量的数据,数据量小的时候没有引擎的优势,数据量大的时候引擎会变得很有优势。ODPS的目标是综合这两者的优点。

PyODPS DataFrame

PyODPS DataFrame是使用Python语言写的,可以使用Python的变量、条件判断、循环。可以使用pandas类似的语法,定义了自己的一套前端,有了更好的表达力。后端可以根据数据来源来决定具体执行的引擎,是visitor的设计模式,可扩展。整个执行是延迟执行,除非用户调用立即执行的方法,否则是不会直接执行的。

从上图中可以看出,语法非常类似于pandas。

表达式和抽象语法树

从上图可以看出,用户从一个原始的Collection来进行GroupBy操作,再进行列选择的操作,最下面是Source的Collection。取了两个字段species,这两个字段是做By操作的,pental_length是进行聚合的操作取聚合值。Species字段是直接取出来,shortest字段是进行加一的操作。

Optimizer(操作合并)

后端首先会使用Optimizer对表达式树进行优化,先做GroupBy,然后在上面做列选择,通过操作合并可以去除petal_length做聚合操作,再加一,最终形成了GroupBy的Collection。

Optimizer(列剪枝)

用户join了两个data frame,再取来自data frame 的两个列的时候,如果提交到一个大数据的环境,这样一个过程是非常低下的,因为不是每个列都用到了。所以要对joined下的列进行剪枝操作。比如,data frame1我们只用到了其中的一个字段,我们只需要将字段截取出来做一个projection来形成新的Collection,data frame2也类似。这样,对这两部分进行校验操作的时候就能极大的减少数据的输出量。

Optimizer(谓词下推)

如果对两个data frame进行joined然后再分别进行过滤的话,这个过滤操作是应该下推到下面来执行的,这样就能减少joined 的输入的量。

可视化

提供了visualize()来方便用户进行可视化。在右边的例子中可以看到,ODSP SQL后端会compile成一条SQL执行。

后端

从上图中可以看出,计算后端是非常灵活的。用户甚至可以joined一个pandas的data frame和maxcompute上一个表的数据。

Analyzer

Analyzer的作用是针对具体的后端,将一些操作进行转化。比如:

有些操作比如value_counts,pandas本身支持,因此对于pandas后端,无需处理;对于ODPS SQL后端,没有一个直接的操作来执行,所以在analyzer执行的时候,会被改写成groupby + sort的操作;

还有一些算子,在compile到ODPS SQL时,没有内建函数能完成,会被改写成自定义函数。

ODPS SQL后端

ODPS SQL后端怎么进行SQL编译再执行的操作?编译器可以从上到下遍历表达式树,找到Join或者Union。对于子过程,进行递归compile。再到Engine来具体执行时,会使用Analyzer对表达式树进行改写,compile自上而下的子过程,自底向上compile成SQL子句,最终得到完整的SQL语句,提交SQL并返回任务。

pandas后端

首先访问这个表达式树,然后对每个表达式树节点对应到pandas操作,整个表达式树遍历完之后就会形成DAG。Engine执行按DAG拓扑顺序执行,不断地把它应用到pandas操作,最终得到一个结果。对于大数据环境来说,pandas后端的作用是做本地DEBUG;当数据量很小时,我们可以使用pandas进行计算。

难点+坑

后端编译出错容易丢失上下文,多次optimize和analyze,导致难以查出是之前哪处visit node导致。解决:保证每个模块独?立性、测试完备;

bytecode兼容问题,maxcompute只支持Python2.7的自定义函数的执行;

SQL的执行顺序。

ML机器学习

机器学习是输入输出一个data frame。比如,有一个iris的data frame,先用name字段来做一个分类字段,调用split方法将其分成60%的训练数据和40%的测试数据。然后初始化一个RandomForests,其里面有一棵决策树,调用train方法训练训练数据,调用predict方法形成一个预测数据,调用segments[0]就可以看到可视化结果。

未来计划

分布式numpy,DataFrame基于分布式numpy的后端;

内存计算,提升交互式体验;

Tensorflow。

下载本文
显示全文
专题