Spark上使用pandas API快速入门

文章最前： 我是Octopus，这个名字来源于我的中文名--章鱼；我热爱编程、热爱算法、热爱开源。所有源码在我的个人github ；这博客是记录我学习的点点滴滴，如果您对 Python、Java、AI、算法有兴趣，可以关注我的动态，一起学习，共同进步。

 这是 Spark 上的 pandas API 的简短介绍，主要面向新用户。本笔记本向您展示 pandas 和 Spark 上的 pandas API 之间的一些关键区别。您可以在快速入门页面的“Live Notebook：Spark 上的 pandas API”中自行运行此示例。

习惯上，我们在Spark上导入pandas API如下：

import pandas as pd
import numpy as np
import pyspark.pandas as ps
from pyspark.sql import SparkSession

对象创建

通过传递值列表来创建 pandas-on-Spark 系列，让 Spark 上的 pandas API 创建默认整数索引：

s = ps.Series([1, 3, 5, np.nan, 6, 8])
s

0    1.0
1    3.0
2    5.0
3    NaN
4    6.0
5    8.0
dtype: float64

通过传递可转换为类似系列的对象字典来创建 pandas-on-Spark DataFrame。

psdf = ps.DataFrame({'a': [1, 2, 3, 4, 5, 6],'b': [100, 200, 300, 400, 500, 600],'c': ["one", "two", "three", "four", "five", "six"]},index=[10, 20, 30, 40, 50, 60])
psdf

a	b	c
10	1	100	one
20	2	200	two
30	3	300	three
40	4	400	four
50	5	500	five
60	6	600	six

创建pandas DataFrame通过numpyt array, 用datetime 作为索引，label列

dates = pd.date_range('20130101', periods=6)
dates

  DatetimeIndex(['2013-01-01', '2013-01-02', '2013-01-03', '2013-01-04', '2013-01-05', '2013-01-06'], dtype='datetime64[ns]', freq='D')

pdf = pd.DataFrame(np.random.randn(6, 4), index=dates, columns=list('ABCD'))
pdf

A	B	C	D
2013-01-01	0.912558	-0.795645	-0.289115	0.187606
2013-01-02	-0.059703	-1.233897	0.316625	-1.226828
2013-01-03	0.332871	-1.262010	-0.434844	-0.579920
2013-01-04	0.924016	-1.022019	-0.405249	-1.036021
2013-01-05	-0.772209	-1.228099	0.068901	0.896679
2013-01-06	1.485582	-0.709306	-0.202637	-0.248766

现在，dataframe能够转换成pandas 在spark上运行

psdf = ps.from_pandas(pdf)
type(psdf)

 pyspark.pandas.frame.DataFrame

看上去和dataframe一样的使用

psdf

A	B	C	D
2013-01-01	0.912558	-0.795645	-0.289115	0.187606
2013-01-02	-0.059703	-1.233897	0.316625	-1.226828
2013-01-03	0.332871	-1.262010	-0.434844	-0.579920
2013-01-04	0.924016	-1.022019	-0.405249	-1.036021
2013-01-05	-0.772209	-1.228099	0.068901	0.896679
2013-01-06	1.485582	-0.709306	-0.202637	-0.248766

当然通过spark pandas dataframe创建pandas  on spark dataframe 非常容易

spark = SparkSession.builder.getOrCreate()
sdf = spark.createDataFrame(pdf)
sdf.show()

 +--------------------+-------------------+--------------------+--------------------+ | A| B| C| D| +--------------------+-------------------+--------------------+--------------------+ | 0.91255803205208|-0.7956452608556638|-0.28911463069772175| 0.18760566615081622| |-0.05970271470242...| -1.233896949308984| 0.3166246451758431| -1.2268284000402265| | 0.33287106947536615|-1.2620100816441786| -0.4348444277082644| -0.5799199651437185| | 0.9240158461589916|-1.0220190956326003| -0.4052488880650239| -1.0360212104348547| | -0.7722090016558953|-1.2280986385313222| 0.0689011451939635| 0.8966790729426755| | 1.4855822995785612|-0.7093056426018517| -0.2026366848847041|-0.24876619876451092| +--------------------+-------------------+--------------------+--------------------+

从 Spark DataFrame 创建 pandas-on-Spark DataFrame。

psdf = sdf.pandas_api()
psdf

A	B	C	D
0	0.912558	-0.795645	-0.289115	0.187606
1	-0.059703	-1.233897	0.316625	-1.226828
2	0.332871	-1.262010	-0.434844	-0.579920
3	0.924016	-1.022019	-0.405249	-1.036021
4	-0.772209	-1.228099	0.068901	0.896679
5	1.485582	-0.709306	-0.202637	-0.248766

具有特定的dtypes。目前支持 Spark 和 pandas 通用的类型。

psdf.dtypes

A    float64
B    float64
C    float64
D    float64
dtype: object

以下是如何显示下面框架中的顶行。

请注意，Spark 数据帧中的数据默认不保留自然顺序。可以通过设置compute.ordered_head选项来保留自然顺序，但它会导致内部排序的性能开销。

psdf.head()

A	B	C	D
0	0.912558	-0.795645	-0.289115	0.187606
1	-0.059703	-1.233897	0.316625	-1.226828
2	0.332871	-1.262010	-0.434844	-0.579920
3	0.924016	-1.022019	-0.405249	-1.036021
4	-0.772209	-1.228099	0.068901	0.896679

 展示index和columns 通过numpy 数据

psdf.index

Int64Index([0, 1, 2, 3, 4, 5], dtype='int64')

psdf.columns

Index(['A', 'B', 'C', 'D'], dtype='object')

psdf.to_numpy()

array([[ 0.91255803, -0.79564526, -0.28911463,  0.18760567],[-0.05970271, -1.23389695,  0.31662465, -1.2268284 ],[ 0.33287107, -1.26201008, -0.43484443, -0.57991997],[ 0.92401585, -1.0220191 , -0.40524889, -1.03602121],[-0.772209  , -1.22809864,  0.06890115,  0.89667907],[ 1.4855823 , -0.70930564, -0.20263668, -0.2487662 ]])

通过简单统计展示你的数据：

psdf.describe()

A	B	C	D
count	6.000000	6.000000	6.000000	6.000000
mean	0.470519	-1.041829	-0.157720	-0.334542
std	0.809428	0.241511	0.294520	0.793014
min	-0.772209	-1.262010	-0.434844	-1.226828
25%	-0.059703	-1.233897	-0.405249	-1.036021
50%	0.332871	-1.228099	-0.289115	-0.579920
75%	0.924016	-0.795645	0.068901	0.187606
max	1.485582	-0.709306	0.316625	0.896679

转置你的数据：

psdf.T

0	1	2	3	4	5
A	0.912558	-0.059703	0.332871	0.924016	-0.772209	1.485582
B	-0.795645	-1.233897	-1.262010	-1.022019	-1.228099	-0.709306
C	-0.289115	0.316625	-0.434844	-0.405249	0.068901	-0.202637
D	0.187606	-1.226828	-0.579920	-1.036021	0.896679	-0.248766

通过index进行排序：

psdf.sort_index(ascending=False)

A	B	C	D
5	1.485582	-0.709306	-0.202637	-0.248766
4	-0.772209	-1.228099	0.068901	0.896679
3	0.924016	-1.022019	-0.405249	-1.036021
2	0.332871	-1.262010	-0.434844	-0.579920
1	-0.059703	-1.233897	0.316625	-1.226828
0	0.912558	-0.795645	-0.289115	0.187606