Python:使用map和reduce写外积
Python: write outer product using map and reduce
假设我有一个矩阵
import numpy as np
from functools import reduce
np.random.seed(123)
X = np.random.normal(size=(5, 2))
我想在不使用 numpy 函数的情况下使用 map
、reduce
和 lambda
函数来计算 X^t X。因为我们可以将 X^t X 写成外积之和,所以我的目标是:
def outer_product(x):
"""Computes outer product of a vector with itself"""
pass
map(outer_product, X)
但是我似乎找不到一种有效的方法来使用 map reduce 编写所有这些内容。
我的尝试
def outer(x):
xxx = np.repeat(x, len(x))
yyy = np.array(list(x) * len(x))
return np.reshape(list(map(lambda x, y: x*y, xxx, yyy)), (len(x), len(x)))
所以
outer(X[0, ])
然后我把协方差矩阵写成如下
def cov(X):
return np.array(reduce(lambda x, y: x + y, list(map(outer, X)))) / np.size(X, 0)
为了回答你的问题,外积可以定义为嵌套映射,例如
outer = lambda V: np.array(list(map(lambda x: list(map(lambda y: x*y, V)), V)))
X = np.random.normal(size=(5, 2))
>>> outer(X[1])
array([[ 0.08007683, -0.42624902], [-0.42624902, 2.26892377]])
其实用列表理解更简单
outer = lambda V: np.array([[x*x1 for a in V] for x1 in V])
会给你同样的结果。然后你可以像
一样映射到你的矩阵
>>> list(map(outer, X))
[array([[ 1.17859381, -1.08274874],
[-1.08274874, 0.99469794]]), array([[ 0.08007683, -0.42624902],
[-0.42624902, 2.26892377]]), array([[ 0.33477825, -0.9555216 ],
[-0.9555216 , 2.72724264]]), array([[5.88877215, 1.04083337],
[1.04083337, 0.18396604]]), array([[ 1.60259461, -1.0972381 ],
[-1.0972381 , 0.75123892]])]
顺便说一下,你的 reduce 部分非常漂亮和简洁。我认为该部分不需要任何进一步的重构。
In [40]: X = np.arange(10).reshape(5,2)
使用已接受答案中的 outer
和 list/map:
In [41]: outer = lambda V: np.array([[x*x1 for x in V] for x1 in V])
In [43]: list(map(outer, X))
Out[43]:
[array([[0, 0],
[0, 1]]),
array([[4, 6],
[6, 9]]),
array([[16, 20],
[20, 25]]),
array([[36, 42],
[42, 49]]),
array([[64, 72],
[72, 81]])]
使用numpy
广播:
In [44]: X[:,:,None]*X[:,None,:]
Out[44]:
array([[[ 0, 0],
[ 0, 1]],
[[ 4, 6],
[ 6, 9]],
[[16, 20],
[20, 25]],
[[36, 42],
[42, 49]],
[[64, 72],
[72, 81]]])
对于 (5,2) X
结果是 (5,2,2) 数组。
一些小测试:
In [55]: timeit list(map(outer, X))
51.8 µs ± 1.29 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
In [57]: timeit [outer(i) for i in X]
49.8 µs ± 65.9 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)
In [58]: timeit X[:,:,None]*X[:,None,:]
5.37 µs ± 11.6 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)
list/map
与列表理解的速度大致相同。一般来说,我发现列表理解更清晰。但是广播的 numpy 操作通常要快得多。
如果要迭代,请考虑使用列表;它通常更快:
In [61]: timeit [outer(i) for i in X.tolist()]
24.3 µs ± 63.6 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)
In [62]: outer0 = lambda V: [[x*x1 for x in V] for x1 in V]
In [64]: timeit [outer0(i) for i in X.tolist()]
7.33 µs ± 16 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)
假设我有一个矩阵
import numpy as np
from functools import reduce
np.random.seed(123)
X = np.random.normal(size=(5, 2))
我想在不使用 numpy 函数的情况下使用 map
、reduce
和 lambda
函数来计算 X^t X。因为我们可以将 X^t X 写成外积之和,所以我的目标是:
def outer_product(x):
"""Computes outer product of a vector with itself"""
pass
map(outer_product, X)
但是我似乎找不到一种有效的方法来使用 map reduce 编写所有这些内容。
我的尝试
def outer(x):
xxx = np.repeat(x, len(x))
yyy = np.array(list(x) * len(x))
return np.reshape(list(map(lambda x, y: x*y, xxx, yyy)), (len(x), len(x)))
所以
outer(X[0, ])
然后我把协方差矩阵写成如下
def cov(X):
return np.array(reduce(lambda x, y: x + y, list(map(outer, X)))) / np.size(X, 0)
为了回答你的问题,外积可以定义为嵌套映射,例如
outer = lambda V: np.array(list(map(lambda x: list(map(lambda y: x*y, V)), V)))
X = np.random.normal(size=(5, 2))
>>> outer(X[1])
array([[ 0.08007683, -0.42624902], [-0.42624902, 2.26892377]])
其实用列表理解更简单
outer = lambda V: np.array([[x*x1 for a in V] for x1 in V])
会给你同样的结果。然后你可以像
一样映射到你的矩阵>>> list(map(outer, X))
[array([[ 1.17859381, -1.08274874],
[-1.08274874, 0.99469794]]), array([[ 0.08007683, -0.42624902],
[-0.42624902, 2.26892377]]), array([[ 0.33477825, -0.9555216 ],
[-0.9555216 , 2.72724264]]), array([[5.88877215, 1.04083337],
[1.04083337, 0.18396604]]), array([[ 1.60259461, -1.0972381 ],
[-1.0972381 , 0.75123892]])]
顺便说一下,你的 reduce 部分非常漂亮和简洁。我认为该部分不需要任何进一步的重构。
In [40]: X = np.arange(10).reshape(5,2)
使用已接受答案中的 outer
和 list/map:
In [41]: outer = lambda V: np.array([[x*x1 for x in V] for x1 in V])
In [43]: list(map(outer, X))
Out[43]:
[array([[0, 0],
[0, 1]]),
array([[4, 6],
[6, 9]]),
array([[16, 20],
[20, 25]]),
array([[36, 42],
[42, 49]]),
array([[64, 72],
[72, 81]])]
使用numpy
广播:
In [44]: X[:,:,None]*X[:,None,:]
Out[44]:
array([[[ 0, 0],
[ 0, 1]],
[[ 4, 6],
[ 6, 9]],
[[16, 20],
[20, 25]],
[[36, 42],
[42, 49]],
[[64, 72],
[72, 81]]])
对于 (5,2) X
结果是 (5,2,2) 数组。
一些小测试:
In [55]: timeit list(map(outer, X))
51.8 µs ± 1.29 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
In [57]: timeit [outer(i) for i in X]
49.8 µs ± 65.9 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)
In [58]: timeit X[:,:,None]*X[:,None,:]
5.37 µs ± 11.6 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)
list/map
与列表理解的速度大致相同。一般来说,我发现列表理解更清晰。但是广播的 numpy 操作通常要快得多。
如果要迭代,请考虑使用列表;它通常更快:
In [61]: timeit [outer(i) for i in X.tolist()]
24.3 µs ± 63.6 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)
In [62]: outer0 = lambda V: [[x*x1 for x in V] for x1 in V]
In [64]: timeit [outer0(i) for i in X.tolist()]
7.33 µs ± 16 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)