【翻译】《利用Python进行数据分析·第2版》第4章（下）NumPy基础：数组和矢量计算

作者：SeanCheney   Python爱好者社区专栏作者

简书专栏：https://www.jianshu.com/u/130f76596b02

前文传送门：

4.3 利用数组进行数据处理

NumPy数组使你可以将许多种数据处理任务表述为简洁的数组表达式（否则需要编写循环）。用数组表达式代替循环的做法，通常被称为矢量化。一般来说，矢量化数组运算要比等价的纯Python方式快上一两个数量级（甚至更多），尤其是各种数值计算。在后面内容中（见附录A）我将介绍广播，这是一种针对矢量化计算的强大手段。

作为简单的例子，假设我们想要在一组值（网格型）上计算函数sqrt(x^2+y^2)。np.meshgrid函数接受两个一维数组，并产生两个二维矩阵（对应于两个数组中所有的(x,y)对）：

In [155]: points = np.arange(-5, 5, 0.01) # 1000 equally spaced points In [156]: xs, ys = np.meshgrid(points, points) In [157]: ys Out[157]: array([[-5.  , -5.  , -5.  , ..., -5.  , -5.  , -5.  ],       [-4.99, -4.99, -4.99, ..., -4.99, -4.99, -4.99],       [-4.98, -4.98, -4.98, ..., -4.98, -4.98, -4.98],       ...,       [ 4.97,  4.97,  4.97, ...,  4.97,  4.97,  4.97],       [ 4.98,  4.98,  4.98, ...,  4.98,  4.98,  4.98],       [ 4.99,  4.99,  4.99, ...,  4.99,  4.99,  4.99]])

现在，对该函数的求值运算就好办了，把这两个数组当做两个浮点数那样编写表达式即可：

In [158]: z = np.sqrt(xs ** 2 + ys ** 2) In [159]: z Out[159]: array([[ 7.0711,  7.064 ,  7.0569, ...,  7.0499,  7.0569,  7.064 ],       [ 7.064 ,  7.0569,  7.0499, ...,  7.0428,  7.0499,  7.0569],       [ 7.0569,  7.0499,  7.0428, ...,  7.0357,  7.0428, 7.0499],       ...,       [ 7.0499,  7.0428,  7.0357, ...,  7.0286,  7.0357,  7.0428],       [ 7.0569,  7.0499,  7.0428, ...,  7.0357,  7.0428,  7.0499],       [ 7.064 ,  7.0569,  7.0499, ...,  7.0428,  7.0499,  7.0569]])

作为第9章的先导，我用matplotlib创建了这个二维数组的可视化：

In [160]: import matplotlib.pyplot as plt In [161]: plt.imshow(z, cmap=plt.cm.gray); plt.colorbar() Out[161]: <matplotlib.colorbar.Colorbar at 0x7f715e3fa630> In [162]: plt.title("Image plot of $\sqrt{x^2 + y^2}$ for a grid of values") Out[162]: <matplotlib.text.Text at 0x7f715d2de748>

见图4-3。这张图是用matplotlib的imshow函数创建的。

图4-3 根据网格对函数求值的结果

将条件逻辑表述为数组运算

numpy.where函数是三元表达式x if condition else y的矢量化版本。假设我们有一个布尔数组和两个值数组：

In [165]: xarr = np.array([1.1, 1.2, 1.3, 1.4, 1.5]) In [166]: yarr = np.array([2.1, 2.2, 2.3, 2.4, 2.5]) In [167]: cond = np.array([True, False, True, True, False])

假设我们想要根据cond中的值选取xarr和yarr的值：当cond中的值为True时，选取xarr的值，否则从yarr中选取。列表推导式的写法应该如下所示：

In [168]: result = [(x if c else y)   .....:           for x, y, c in zip(xarr, yarr, cond)] In [169]: result Out[169]: [1.1000000000000001, 2.2000000000000002, 1.3, 1.3999999999999999, 2.5]

这有几个问题。第一，它对大数组的处理速度不是很快（因为所有工作都是由纯Python完成的）。第二，无法用于多维数组。若使用np.where，则可以将该功能写得非常简洁：

In [170]: result = np.where(cond, xarr, yarr) In [171]: result Out[171]: array([ 1.1,  2.2,  1.3,  1.4,  2.5])

np.where的第二个和第三个参数不必是数组，它们都可以是标量值。在数据分析工作中，where通常用于根据另一个数组而产生一个新的数组。假设有一个由随机数据组成的矩阵，你希望将所有正值替换为2，将所有负值替换为－2。若利用np.where，则会非常简单：

In [172]: arr = np.random.randn(4, 4) In [173]: arr Out[173]: array([[-0.5031, -0.6223, -0.9212, -0.7262],       [ 0.2229,  0.0513, -1.1577,  0.8167],       [ 0.4336,  1.0107,  1.8249, -0.9975],       [ 0.8506, -0.1316,  0.9124,  0.1882]]) In [174]: arr > 0 Out[174]: array([[False, False, False, False],       [ True,  True, False,  True],       [ True,  True,  True, False],       [ True, False,  True,  True]], dtype=bool) In [175]: np.where(arr > 0, 2, -2) Out[175]: array([[-2, -2, -2, -2],       [ 2,  2, -2,  2],       [ 2,  2,  2, -2],       [ 2, -2,  2,  2]])

使用np.where，可以将标量和数组结合起来。例如，我可用常数2替换arr中所有正的值：

In [176]: np.where(arr > 0, 2, arr) # set only positive values to 2 Out[176]: array([[-0.5031, -0.6223, -0.9212, -0.7262],       [ 2.    ,  2.    , -1.1577,  2.    ],       [ 2.    ,  2.    ,  2.    , -0.9975],       [ 2.    , -0.1316,  2.    ,  2.    ]])

传递给where的数组大小可以不相等，甚至可以是标量值。

数学和统计方法

可以通过数组上的一组数学函数对整个数组或某个轴向的数据进行统计计算。sum、mean以及标准差std等聚合计算（aggregation，通常叫做约简（reduction））既可以当做数组的实例方法调用，也可以当做顶级NumPy函数使用。

这里，我生成了一些正态分布随机数据，然后做了聚类统计：

In [177]: arr = np.random.randn(5, 4) In [178]: arr Out[178]: array([[ 2.1695, -0.1149,  2.0037,  0.0296],       [ 0.7953,  0.1181, -0.7485,  0.585 ],       [ 0.1527, -1.5657, -0.5625, -0.0327],       [-0.929 , -0.4826, -0.0363,  1.0954],       [ 0.9809, -0.5895,  1.5817, -0.5287]]) In [179]: arr.mean() Out[179]: 0.19607051119998253 In [180]: np.mean(arr) Out[180]: 0.19607051119998253 In [181]: arr.sum() Out[181]: 3.9214102239996507

mean和sum这类的函数可以接受一个axis选项参数，用于计算该轴向上的统计值，最终结果是一个少一维的数组：

In [182]: arr.mean(axis=1) Out[182]: array([ 1.022 ,  0.1875, -0.502 , -0.0881,  0.3611]) In [183]: arr.sum(axis=0) Out[183]: array([ 3.1693, -2.6345,  2.2381,  1.1486])

这里，arr.mean(1)是“计算行的平均值”，arr.sum(0)是“计算每列的和”。

其他如cumsum和cumprod之类的方法则不聚合，而是产生一个由中间结果组成的数组：

In [184]: arr = np.array([0, 1, 2, 3, 4, 5, 6, 7]) In [185]: arr.cumsum() Out[185]: array([ 0,  1,  3,  6, 10, 15, 21, 28])

在多维数组中，累加函数（如cumsum）返回的是同样大小的数组，但是会根据每个低维的切片沿着标记轴计算部分聚类：

In [186]: arr = np.array([[0, 1, 2], [3, 4, 5], [6, 7, 8]]) In [187]: arr Out[187]: array([[0, 1, 2],       [3, 4, 5],       [6, 7, 8]]) In [188]: arr.cumsum(axis=0) Out[188]: array([[ 0,  1,  2],       [ 3,  5,  7],       [ 9, 12, 15]]) In [189]: arr.cumprod(axis=1) Out[189]: array([[  0,   0,   0],       [  3,  12,  60],       [  6,  42, 336]])

表4-5列出了全部的基本数组统计方法。后续章节中有很多例子都会用到这些方法。

用于布尔型数组的方法

在上面这些方法中，布尔值会被强制转换为1（True）和0（False）。因此，sum经常被用来对布尔型数组中的True值计数：

In [190]: arr = np.random.randn(100) In [191]: (arr > 0).sum() # Number of positive values Out[191]: 42

另外还有两个方法any和all，它们对布尔型数组非常有用。any用于测试数组中是否存在一个或多个True，而all则检查数组中所有值是否都是True：

In [192]: bools = np.array([False, False, True, False]) In [193]: bools.any() Out[193]: True In [194]: bools.all() Out[194]: False

这两个方法也能用于非布尔型数组，所有非0元素将会被当做True。

排序

跟Python内置的列表类型一样，NumPy数组也可以通过sort方法就地排序：

In [195]: arr = np.random.randn(6) In [196]: arr Out[196]: array([ 0.6095, -0.4938,  1.24  , -0.1357,  1.43  , -0.8469]) In [197]: arr.sort() In [198]: arr Out[198]: array([-0.8469, -0.4938, -0.1357,  0.6095,  1.24  ,  1.43  ])

多维数组可以在任何一个轴向上进行排序，只需将轴编号传给sort即可：

In [199]: arr = np.random.randn(5, 3) In [200]: arr Out[200]: array([[ 0.6033,  1.2636, -0.2555],       [-0.4457,  0.4684, -0.9616],       [-1.8245,  0.6254,  1.0229],       [ 1.1074,  0.0909, -0.3501],       [ 0.218 , -0.8948, -1.7415]]) In [201]: arr.sort(1) In [202]: arr Out[202]: array([[-0.2555,  0.6033,  1.2636],       [-0.9616, -0.4457,  0.4684],       [-1.8245,  0.6254,  1.0229],       [-0.3501,  0.0909,  1.1074],       [-1.7415, -0.8948,  0.218 ]])

顶级方法np.sort返回的是数组的已排序副本，而就地排序则会修改数组本身。计算数组分位数最简单的办法是对其进行排序，然后选取特定位置的值：

In [203]: large_arr = np.random.randn(1000) In [204]: large_arr.sort() In [205]: large_arr[int(0.05 * len(large_arr))] # 5% quantile Out[205]: -1.5311513550102103

更多关于NumPy排序方法以及诸如间接排序之类的高级技术，请参阅附录A。在pandas中还可以找到一些其他跟排序有关的数据操作（比如根据一列或多列对表格型数据进行排序）。

唯一化以及其它的集合逻辑

NumPy提供了一些针对一维ndarray的基本集合运算。最常用的可能要数np.unique了，它用于找出数组中的唯一值并返回已排序的结果：

In [206]: names = np.array(['Bob', 'Joe', 'Will', 'Bob', 'Will', 'Joe', 'Joe']) In [207]: np.unique(names) Out[207]: array(['Bob', 'Joe', 'Will'],      dtype='<U4') In [208]: ints = np.array([3, 3, 3, 2, 2, 1, 1, 4, 4]) In [209]: np.unique(ints) Out[209]: array([1, 2, 3, 4])

拿跟np.unique等价的纯Python代码来对比一下：

In [210]: sorted(set(names)) Out[210]: ['Bob', 'Joe', 'Will']

另一个函数np.in1d用于测试一个数组中的值在另一个数组中的成员资格，返回一个布尔型数组：

In [211]: values = np.array([6, 0, 0, 3, 2, 5, 6]) In [212]: np.in1d(values, [2, 3, 6]) Out[212]: array([ True, False, False,  True,  True, False,  True], dtype=bool)

NumPy中的集合函数请参见表4-6。

4.4 用于数组的文件输入输出

NumPy能够读写磁盘上的文本数据或二进制数据。这一小节只讨论NumPy的内置二进制格式，因为更多的用户会使用pandas或其它工具加载文本或表格数据（见第6章）。

np.save和np.load是读写磁盘数组数据的两个主要函数。默认情况下，数组是以未压缩的原始二进制格式保存在扩展名为.npy的文件中的：

In [213]: arr = np.arange(10) In [214]: np.save('some_array', arr)

如果文件路径末尾没有扩展名.npy，则该扩展名会被自动加上。然后就可以通过np.load读取磁盘上的数组：

In [215]: np.load('some_array.npy') Out[215]: array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

通过np.savez可以将多个数组保存到一个未压缩文件中，将数组以关键字参数的形式传入即可：

In [216]: np.savez('array_archive.npz', a=arr, b=arr)

加载.npz文件时，你会得到一个类似字典的对象，该对象会对各个数组进行延迟加载：

In [217]: arch = np.load('array_archive.npz') In [218]: arch['b'] Out[218]: array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

如果数据压缩的很好，就可以使用numpy.savez_compressed：

In [219]: np.savez_compressed('arrays_compressed.npz', a=arr, b=arr)

4.5 线性代数

线性代数（如矩阵乘法、矩阵分解、行列式以及其他方阵数学等）是任何数组库的重要组成部分。不像某些语言（如MATLAB），通过*对两个二维数组相乘得到的是一个元素级的积，而不是一个矩阵点积。因此，NumPy提供了一个用于矩阵乘法的dot函数（既是一个数组方法也是numpy命名空间中的一个函数）：

In [223]: x = np.array([[1., 2., 3.], [4., 5., 6.]]) In [224]: y = np.array([[6., 23.], [-1, 7], [8, 9]]) In [225]: x Out[225]: array([[ 1.,  2.,  3.],       [ 4.,  5.,  6.]]) In [226]: y Out[226]: array([[  6.,  23.],       [ -1.,   7.],       [  8.,   9.]]) In [227]: x.dot(y) Out[227]: array([[  28.,   64.],       [  67.,  181.]])

x.dot(y)等价于np.dot(x, y)：

In [228]: np.dot(x, y) Out[228]: array([[  28.,   64.],       [  67.,  181.]])

一个二维数组跟一个大小合适的一维数组的矩阵点积运算之后将会得到一个一维数组：

In [229]: np.dot(x, np.ones(3)) Out[229]: array([  6.,  15.])

@符（类似Python 3.5）也可以用作中缀运算符，进行矩阵乘法：

In [230]: x @ np.ones(3) Out[230]: array([  6.,  15.])

numpy.linalg中有一组标准的矩阵分解运算以及诸如求逆和行列式之类的东西。它们跟MATLAB和R等语言所使用的是相同的行业标准线性代数库，如BLAS、LAPACK、Intel MKL（Math Kernel Library，可能有，取决于你的NumPy版本）等：

In [231]: from numpy.linalg import inv, qr In [232]: X = np.random.randn(5, 5) In [233]: mat = X.T.dot(X) In [234]: inv(mat) Out[234]: array([[  933.1189,   871.8258, -1417.6902, -1460.4005,  1782.1391],       [  871.8258,   815.3929, -1325.9965, -1365.9242,  1666.9347],       [-1417.6902, -1325.9965,  2158.4424,  2222.0191, -2711.6822],       [-1460.4005, -1365.9242,  2222.0191,  2289.0575, -2793.422 ],       [ 1782.1391,  1666.9347, -2711.6822, -2793.422 ,  3409.5128]]) In [235]: mat.dot(inv(mat)) Out[235]: array([[ 1.,  0., -0., -0., -0.],       [-0.,  1.,  0.,  0.,  0.],       [ 0.,  0.,  1.,  0.,  0.],       [-0.,  0.,  0.,  1., -0.],       [-0.,  0.,  0.,  0.,  1.]]) In [236]: q, r = qr(mat) In [237]: r Out[237]: array([[-1.6914,  4.38  ,  0.1757,  0.4075, -0.7838],       [ 0.    , -2.6436,  0.1939, -3.072 , -1.0702],       [ 0.    ,  0.    , -0.8138,  1.5414,  0.6155],       [ 0.    ,  0.    ,  0.    , -2.6445, -2.1669],       [ 0.    ,  0.    ,  0.    ,  0.    ,  0.0002]])

表达式X.T.dot(X)计算X和它的转置X.T的点积。

表4-7中列出了一些最常用的线性代数函数。

4.6 伪随机数生成

numpy.random模块对Python内置的random进行了补充，增加了一些用于高效生成多种概率分布的样本值的函数。例如，你可以用normal来得到一个标准正态分布的4×4样本数组：

In [238]: samples = np.random.normal(size=(4, 4)) In [239]: samples Out[239]: array([[ 0.5732,  0.1933,  0.4429,  1.2796],       [ 0.575 ,  0.4339, -0.7658, -1.237 ],       [-0.5367,  1.8545, -0.92  , -0.1082],       [ 0.1525,  0.9435, -1.0953, -0.144 ]])

而Python内置的random模块则只能一次生成一个样本值。从下面的测试结果中可以看出，如果需要产生大量样本值，numpy.random快了不止一个数量级：

In [240]: from random import normalvariate In [241]: N = 1000000 In [242]: %timeit samples = [normalvariate(0, 1) for _ in range(N)] 1.77 s +- 126 ms per loop (mean +- std. dev. of 7 runs, 1 loop each) In [243]: %timeit np.random.normal(size=N) 61.7 ms +- 1.32 ms per loop (mean +- std. dev. of 7 runs, 10 loops each)

我们说这些都是伪随机数，是因为它们都是通过算法基于随机数生成器种子，在确定性的条件下生成的。你可以用NumPy的np.random.seed更改随机数生成种子：

In [244]: np.random.seed(1234)

numpy.random的数据生成函数使用了全局的随机种子。要避免全局状态，你可以使用numpy.random.RandomState，创建一个与其它隔离的随机数生成器：

In [245]: rng = np.random.RandomState(1234) In [246]: rng.randn(10) Out[246]: array([ 0.4714, -1.191 ,  1.4327, -0.3127, -0.7206,  0.8872,  0.8596,       -0.6365,  0.0157, -2.2427])

表4-8列出了numpy.random中的部分函数。在下一节中，我将给出一些利用这些函数一次性生成大量样本值的范例。

4.7 示例：随机漫步

我们通过模拟随机漫步来说明如何运用数组运算。先来看一个简单的随机漫步的例子：从0开始，步长1和－1出现的概率相等。

下面是一个通过内置的random模块以纯Python的方式实现1000步的随机漫步：

In [247]: import random   .....: position = 0   .....: walk = [position]   .....: steps = 1000   .....: for i in range(steps):   .....:     step = 1 if random.randint(0, 1) else -1   .....:     position += step   .....:     walk.append(position)   .....:

图4-4是根据前100个随机漫步值生成的折线图：

In [249]: plt.plot(walk[:100])

图4-4 简单的随机漫步

不难看出，这其实就是随机漫步中各步的累计和，可以用一个数组运算来实现。因此，我用np.random模块一次性随机产生1000个“掷硬币”结果（即两个数中任选一个），将其分别设置为1或－1，然后计算累计和：

In [251]: nsteps = 1000 In [252]: draws = np.random.randint(0, 2, size=nsteps) In [253]: steps = np.where(draws > 0, 1, -1) In [254]: walk = steps.cumsum()

有了这些数据之后，我们就可以沿着漫步路径做一些统计工作了，比如求取最大值和最小值：

In [255]: walk.min() Out[255]: -3 In [256]: walk.max() Out[256]: 31

现在来看一个复杂点的统计任务——首次穿越时间，即随机漫步过程中第一次到达某个特定值的时间。假设我们想要知道本次随机漫步需要多久才能距离初始0点至少10步远（任一方向均可）。np.abs(walk)>=10可以得到一个布尔型数组，它表示的是距离是否达到或超过10，而我们想要知道的是第一个10或－10的索引。可以用argmax来解决这个问题，它返回的是该布尔型数组第一个最大值的索引（True就是最大值）：

In [257]: (np.abs(walk) >= 10).argmax() Out[257]: 37

注意，这里使用argmax并不是很高效，因为它无论如何都会对数组进行完全扫描。在本例中，只要发现了一个True，那我们就知道它是个最大值了。

一次模拟多个随机漫步

如果你希望模拟多个随机漫步过程（比如5000个），只需对上面的代码做一点点修改即可生成所有的随机漫步过程。只要给numpy.random的函数传入一个二元元组就可以产生一个二维数组，然后我们就可以一次性计算5000个随机漫步过程（一行一个）的累计和了：

In [258]: nwalks = 5000 In [259]: nsteps = 1000 In [260]: draws = np.random.randint(0, 2, size=(nwalks, nsteps)) # 0 or 1 In [261]: steps = np.where(draws > 0, 1, -1) In [262]: walks = steps.cumsum(1) In [263]: walks Out[263]: array([[  1,   0,   1, ...,   8,   7,   8],       [  1,   0,  -1, ...,  34,  33,  32],       [  1,   0,  -1, ...,   4,   5,   4],       ...,       [  1,   2,   1, ...,  24,  25,  26],       [  1,   2,   3, ...,  14,  13,  14],       [ -1,  -2,  -3, ..., -24, -23, -22]])

现在，我们来计算所有随机漫步过程的最大值和最小值：

In [264]: walks.max() Out[264]: 138 In [265]: walks.min() Out[265]: -133

得到这些数据之后，我们来计算30或－30的最小穿越时间。这里稍微复杂些，因为不是5000个过程都到达了30。我们可以用any方法来对此进行检查：

In [266]: hits30 = (np.abs(walks) >= 30).any(1) In [267]: hits30 Out[267]: array([False,  True, False, ..., False,  True, False], dtype=bool) In [268]: hits30.sum() # Number that hit 30 or -30 Out[268]: 3410

然后我们利用这个布尔型数组选出那些穿越了30（绝对值）的随机漫步（行），并调用argmax在轴1上获取穿越时间：

In [269]: crossing_times = (np.abs(walks[hits30]) >= 30).argmax(1) In [270]: crossing_times.mean() Out[270]: 498.88973607038122

请尝试用其他分布方式得到漫步数据。只需使用不同的随机数生成函数即可，如normal用于生成指定均值和标准差的正态分布数据：

In [271]: steps = np.random.normal(loc=0, scale=0.25,   .....:                          size=(nwalks, nsteps))

4.8 结论

虽然本书剩下的章节大部分是用pandas规整数据，我们还是会用到相似的基于数组的计算。在附录A中，我们会深入挖掘NumPy的特点，进一步学习数组的技巧。

赞赏作者

Python爱好者社区历史文章大合集：

福利：文末扫码立刻关注公众号，“Python爱好者社区”，开始学习Python课程：

关注后在公众号内回复“课程”即可获取：

小编的Python入门视频课程！！！

崔老师爬虫实战案例免费学习视频。

丘老师数据科学入门指导免费学习视频。

陈老师数据分析报告制作免费学习视频。

玩转大数据分析！Spark2.X+Python 精华实战课程免费学习视频。

丘老师Python网络爬虫实战免费学习视频。