什么是PyODPS DataFrame

最近已经写了几篇有关PyODPS DataFrame的文章，但是还是有些同学不明白PyODPS DataFrame是什么，以及能做什么事情。这篇文章，我会做出解释，以及简单介绍一下实现的原理。

PyODPS DataFrame

首先什么是DataFrame，我在以前的文章也解释过，我们可以把它认为是二维表结构。在单机上，数据科学家一般会使用R或者Python库pandas来做数据分析，DataFrame是它们上广泛使用的数据结构。在DataFrame上，我们可以做过滤、列筛选、join、union等等操作。
因此，DataFrame也常常拿来和SQL做比较。我觉得主要的区别有：

可能每个系统都有自己的SQL语法，但是对于DataFrame来说，可以把一套语法应用到不同的系统中，也就是说，各个系统上层的DataFrame语法可以是一致的。

DataFrame可以和本身的实现语言相关，因此能用到语言相关的特性，变量赋值、和语言三方库集成等等都不在话下。

因此从第一点上来说，就能解释为什么我们的PyODPS DataFrame能在ODPS和本地上执行了。同样的语法，灵活性很高。
对于PyODPS DataFrame来说，什么时候数据在MaxCompute上执行，什么时候在本地执行呢？这和用户的初始输入有关。当用户用MaxCompute表来初始化DataFrame的时候，后续计算就会在MaxCompute上执行。

In [5]: iris = DataFrame(o.get_table('pyodps_iris'))

In [6]: iris[iris.sepalwidth < 4].head(3)
|==========================================|   1 /  1  (100.00%) 19s
Out[6]:
sepallength  sepalwidth  petallength  petalwidth         name
0          5.1         3.5          1.4         0.2  Iris-setosa
1          4.9         3.0          1.4         0.2  Iris-setosa
2          4.7         3.2          1.3         0.2  Iris-setosa

比如例子里，我们用一张MaxCompute表来初始化，因此后续的计算都是在MaxCompute上执行的。而如果用pandas DataFrame来初始化，那么后续的计算就是在本地执行。
我们前面一篇文章提过，我们从0.4版本开始带来一个特性，我们能join SQL和本地数据时，具体是怎样实现的呢？其实很简单，我们先把本地数据都计算完成，然后通过Tunnel上传到ODPS，再执行ODPS上的计算。
在ODPS上计算时，和本地的计算能力是无关的，除非获取最终计算结果，数据也不会放在本地。
对于在ODPS上的计算，目前来说，我们绝大多数的操作会使用ODPS SQL执行，但有部分情况，我们会使用tunnel执行，以提高执行速度。这些包括：

对原始表筛选字段

非分区表的切片，或分区表不选取分区或者选取前几个分区字段的切片

非分区表取条数总数，或分区表选取分区的条数总数

举个例子，我们的pyodps_iris是个非分区表，以下情况会使用tunnel，而不是转化成SQL执行。

In [7]: iris.count()
|==========================================|   1 /  1  (100.00%) 0s
150

In [10]: iris.exclude('name')[:3]
|==========================================|   1 /  1  (100.00%) 0s

sepallength  sepalwidth  petallength  petalwidth
0          5.1         3.5          1.4         0.2
1          4.9         3.0          1.4         0.2
2          4.7         3.2          1.3         0.2

可以看到，使用Tunnel的计算是很快的。因此，我们可以利用这个特性来从ODPS上下载少量数据，来利用本地计算来做debug。

In [6]: iris[iris.sepalwidth < 4].head(3)  # 利用ODPS计算时，对小数据量是没有优势的
|==========================================|   1 /  1  (100.00%) 19s
Out[6]:
sepallength  sepalwidth  petallength  petalwidth         name
0          5.1         3.5          1.4         0.2  Iris-setosa
1          4.9         3.0          1.4         0.2  Iris-setosa
2          4.7         3.2          1.3         0.2  Iris-setosa

In [11]: local_iris = iris[:100].to_pandas(wrap=True)
|==========================================|   1 /  1  (100.00%) 0s

In [12]: local_iris[local_iris.sepalwidth < 4].head(3)
|==========================================|   1 /  1  (100.00%) 0s
Out[12]:
sepallength  sepalwidth  petallength  petalwidth         name
0          5.1         3.5          1.4         0.2  Iris-setosa
1          4.9         3.0          1.4         0.2  Iris-setosa
2          4.7         3.2          1.3         0.2  Iris-setosa

wrap为True时，等同于

DataFrame(iris[:100].to_pandas())

。

原理简述

下面，简单来说下PyODPS DataFrame的计算原理。
在某种意义上，PyODPS DataFrame可以认为是DSL（领域特定语言）。在到立即执行的操作（如execute）前，得到的都是一个AST（抽象语法树）。
在交互式环境下，为了方便，我们在repr一个对象时，里面会调用立即执行的方法。因此，我们先把这个选项关掉，来看看执行后会是什么。

In [13]: options.interactive = False

In [14]: iris[iris.sepalwidth < 4][:10]
Out[14]:
Collection: ref_0
odps.Table
name: odps_test_sqltask_finance.`pyodps_iris`
schema:
sepallength : double
sepalwidth : double
petallength : double
petalwidth : double
name : string

Collection: ref_1
Filter[collection]
collection: ref_0
predicate:
Less[sequence(boolean)]
sepalwidth = Column[sequence(float64)] 'sepalwidth' from collection ref_0
Scalar[int8]
4

Slice[collection]
collection: ref_1
stop:
Scalar[int8]
10

现在我们把verbose打开，执行的中间过程会被打印出来，我们可以看到在ODPS上，目前会把这个AST给compile成ODPS SQL来执行。

In [15]: options.verbose = True

In [16]: iris[iris.sepalwidth < 4][:10].execute()

Sql compiled:
CREATE TABLE tmp_pyodps_07ec2ed0_88c5_4649_9413_0bce14f72d6f LIFECYCLE 1 AS
SELECT *
FROM odps_test_sqltask_finance.`pyodps_iris` t1
WHERE t1.`sepalwidth` < 4 LIMIT 10 logview: http://webconsole.odps.aliyun-inc.com:8080/logview/?*** |==========================================| 1 / 1 (100.00%) 32s Out[16]: sepallength sepalwidth petallength petalwidth name 0 5.1 3.5 1.4 0.2 Iris-setosa 1 4.9 3.0 1.4 0.2 Iris-setosa 2 4.7 3.2 1.3 0.2 Iris-setosa 3 4.6 3.1 1.5 0.2 Iris-setosa 4 5.0 3.6 1.4 0.2 Iris-setosa 5 5.4 3.9 1.7 0.4 Iris-setosa 6 4.6 3.4 1.4 0.3 Iris-setosa 7 5.0 3.4 1.5 0.2 Iris-setosa 8 4.4 2.9 1.4 0.2 Iris-setosa 9 4.9 3.1 1.5 0.1 Iris-setosa

而对于本地数据，我们在compile阶段会把AST转化成一个执行DAG（有向无环图），在执行阶段，会按照DAG的拓扑顺序来执行，得到最终结果。

In [17]: local_iris[local_iris.sepalwidth < 4][:10].compile() Out[17]: <odps.dag.DAG at 0x10c233950>

好了，至此，已经简单说明了PyODPS DataFrame框架的执行原理。
PyODPS还很年轻，期待大家来使用、提feature、贡献代码。

安装方法：pip install pyodps

Github：https://github.com/aliyun/aliyun-odps-python-sdk

外部文档：http://pyodps.readthedocs.io/

bug report：https://github.com/aliyun/aliyun-odps-python-sdk/issues