【问题标题】:Vectorize a Newton method in Python/Numpy在 Python/Numpy 中矢量化牛顿方法
【发布时间】:2022-01-05 07:39:18
【问题描述】:

我正在尝试确定 Python/Numpy 是否是开发我的数字软件的可行替代方案,该软件已经在 C++ 中可用。为了在 Python/Numpy 中获得性能,需要“矢量化”代码。但事实证明,一旦我离开非常简单的示例,我就很难对代码进行矢量化(我不是在谈论 SIMD 指令,而是在谈论没有循环的“高效 Numpy 代码”)。这是我想在 Python/Numpy 中高效实现的算法。

  1. 创建一个 numpy 数组,其中包含:1.0、1.0 + 1/n、1.0 + 2/n、...、2.0
  2. 对于数组中的每个 u,使用牛顿法计算 x^2 - u 的根,当 |dx| 时停止
  3. 对结果数组的所有元素求和

这是我要加速的 Python 算法

import numpy as np

n = 1000000
data = np.arange(1.0, 2.0, 1.0 / n)

def newton(u):
  x = 2.0
  while True:
    f = x**2 - u
    df_dx = 2 * x
    dx = f / df_dx
    if (abs(dx) <= 1.0e-7):
      break
    x -= dx
  return x

  result = map(newton, data)

  print result[n - 1]

这里是 C++11 中算法的一个版本

#include <iostream>
#include <vector>
#include <cmath>

int main (int argc, char const *argv[]) {
  auto n = std::size_t{100000000};

  auto v = std::vector<double>(n + 1);
  for(size_t k = 0; k < v.size(); ++k) {
    v[k] = 1.0 + static_cast<double>(k) / n;
  }

  auto result = std::vector<double>(n + 1);
  for(size_t k = 0; k < v.size(); ++k) {
    auto x = double{2.0};
    while(true) {
      auto f = double{x * x - v[k]};
      auto df_dx = double{2 * x};
      auto dx = double{f / df_dx};
      if (std::abs(dx) <= 1.0e-7) {
        break;
      }
      x -= dx;
    }
    result[k] = x;
  }

  auto somme = double{0.0};
  for(size_t k = 0; k < result.size(); ++k) {
    somme += result[k];
  }

  std::cout << somme << std::endl;
  return 0;
}

在我的机器上运行需要 2.9 秒。有没有办法制作一个快速的 Python/Numpy 算法来做同样的事情(我愿意得到慢不到 5 倍的东西)。

谢谢。

【问题讨论】:

  • numpy 的好处是它在幕后进行矢量化,消除了很多(但不是全部)你在编写 C++ 时需要做的优化。当然,您肯定需要使用 numpy 的所有工具来获得最高性能的代码,例如使用 ndarrays 代替列表列表。
  • @MattDMo:我正在寻找一个实际有效的代码。我只是不知道该怎么做,这就是我寻求帮助的原因。
  • '向量化'对于串行操作很困难 - 步骤 i 中的计算取决于步骤 i-1 的结果。当所有步骤都可以并行完成时会更容易 - 即评估顺序无关紧要。
  • @hpaulj:这里不是这样,因为 newton(u1) 和 newton(u2) 是完全独立的。只是在 Python/Numpy 中没有办法有效地做到这一点。我发现的唯一解决方案是使用 Cython 或 Numba,这对我来说是不行的。我不想为这么简单的事情“破解”。
  • newton 循环中,x 一步依赖于上一步的xx -= dxdx 本身是x 的函数)。这就是我所说的串行或迭代解决方案的意思。

标签: python numpy


【解决方案1】:

您可以使用 numpy 高效地执行第 1 步:

1.0 + np.arange(n + 1) / n

但是我认为您需要 np.vectorize() 方法将 x 反馈到您的计算值中,并且它不是一个有效的函数(基本上是 python 循环的包装器)。如果你可以使用 scipy,那么内置的方法可能会做你想做的事情http://docs.scipy.org/doc/scipy-0.14.0/reference/generated/scipy.optimize.newton.html

编辑:在考虑了更多之后,我跟进了@ev-br 的观点并尝试了一些替代方案。屏蔽使用了太多的处理,但 abs().max() 非常快,因此折衷方案可能是在数组的第一维和迭代方向上“将问题分成块”。在我相当低功耗的笔记本电脑上,以下操作并不算太糟糕(

n = 100000000
m = 5000000

block = 3
u = 1.0 + np.arange(n + 1) / n
x = np.full(u.shape, 2.0)
dx = np.ones(u.shape)

for i in range(0, n, m):
  while np.abs(dx[i:i+m]).max() > 1.0e-7:
    for j in range(block):
      dx[i:i+m] = (x[i:i+m] ** 2 - u[i:i+m]) / (2 * x[i:i+m])
      x[i:i+m] -= dx[i:i+m]

【讨论】:

  • 瓶颈是牛顿方法,scipy.optimize.newton 无济于事,因为牛顿方法停止标准将根据你的不同迭代次数停止。
  • 是的 newton 可能不是正确的函数,您可以使用 optimize.fsolve(func, np.full(data.shape, 2.0), data) 但它比简单地将函数放入 np.vectorize() 慢很多,而且比 c++ 慢很多。大概您不想真正使用数值方法解决 x**2 ,这只是具有非平凡根的事物的占位符。否则你可以做(1.0 + np.arange(n + 1) / n) ** 0.5
【解决方案2】:

这是一个玩具示例。请注意,向量化通常意味着编写代码,就好像你在处理数字一样,让 numpy 发挥它的魔力:

>>> import numpy as np
>>> a = np.array([1., 2., 3.])
>>> def f(x):
...    return x**2 - a, 2.*x    # function and derivative
>>>
>>> def newt(f, x0):
...    x = np.asarray(x0)
...    for _ in range(5):    # hardcode the number of iterations (I know)
...        v, dv = f(x)
...        x -=  v / dv
...    return x
>>> 
>>> newt(f, [1., 1., 1.])
array([ 1.        ,  1.41421356,  1.73205081])

如果这是一个性能瓶颈,这不太可能与手写的 C++ 代码竞争:首先,您正在使用所有开销来操作 python 对象;那么 numpy 可能会在后台进行一系列数组分配。

一个通常可行的策略是首先在 python/numpy 中编写内容,然后将瓶颈转移到已编译的代码中——例如 Cython 或由 Cython 包装的 C++。在这种特殊情况下,由于您已经拥有 C++ 代码,因此仅使用 Cython 包装它可能是最简单的,但 YMMV。

【讨论】:

  • ev-br:在我想要的代码中,牛顿法的迭代次数取决于你。这里,迭代次数是全局的。
  • @InsideLoop:使用矢量化方法可以做的最好的事情是类似于while (dx &lt; eps).all(),其中dxeps 都是数组。这确实会迭代 所有 组件,只要其中 任何一个 组件尚未收敛。
  • 其实只能通过布尔数组mask = dx &lt; eps:x[mask] -= v[mask] / dv[mask]来迭代未收敛的组件。
【解决方案3】:

我不希望将小 sn-ps 代码作为解决方案,但这里有一些东西可以帮助您入门。我强烈怀疑您只是在 python 中声明这样一个数组而没有花费太多时间在上面时遇到麻烦,所以我主要会帮助您。

就平方根而言,请添加您的示例 python 代码,然后我会看看我可以帮助优化什么。在我的示例中,根和总和是使用默认的 numpy 函数/方法找到的。

def summing():
    n = 1000000
    ar = np.arange(0, n)
    ar = ar/float(n)
    ar = ar + np.ones(n)
    sqrt = np.sqrt(ar)
    return np.sum(ar)

简而言之,要获得起始数组,最好使用“解决方法”。

  • 使用值 `[1,2,3,....n] 初始化数组 ar
  • arn 分开。这让我们成为1/n, 2/n ... 成员
  • 添加一个相同维度的数组,只包含数字1.0 这为我们提供了我们所追求的完整数组[ 1., 1.000001, 1.000002, ..., 1.999998, 1.999999])。如果我没听错的话。
  • 求平方根,求和

10 个连续执行时间的平均值为 0.018786 秒。

【讨论】:

  • 我的代码是真实代码的“短版”,我试图反演的函数没有封闭形式的反函数。
【解决方案4】:

显然我迟到了 6 年,但这个问题是人们有效使用 numpy 进行真正科学工作的常见绊脚石。 @ev-br 的答案涵盖了基本思想。 OP 指出,那里提供的解决方案(甚至修改为在满足收敛标准时停止迭代,而不是在固定次数的迭代之后)对 u 的每个元素采用相同数量的传递。我想展示如何使用纯 numpy 代码来避免这种反对意见,并在 @ev-br 的评论中明确提出掩码建议。

但是,我还想指出,在许多现实世界的情况下,牛顿式迭代收敛的次数变化很小,以至于我在这里说明的这种通用技术实际上会显着减慢 numpy 代码。如果平均迭代次数在最大迭代次数的两倍或三倍之内,您应该坚持使用更接近@ev-br 的答案(包括他的第一条评论)。

您需要了解的 numpy 性能数字如下: 在纯 numpy 代码中循环遍历数组索引将比在编译代码中慢 200 到 500 倍。另一方面,如果你设法使用 numpy 的数组语法来避免所有索引循环,你可以得到大约 5 倍的编译速度。 (因子 5 部分是因为@ev-br 提到的内存管理,但也因为优化的编译代码在每个索引循环内重叠了许多不同的算术运算,而 numpy 只执行一个算术运算,每次之后将所有内容存储回内存操作。)关键是100倍的差异意味着在numpy代码中做大量的“额外”工作通常是值得的:即使你在向量化的numpy代码中做浮点运算的10倍,它仍然会运行比避免“额外”工作的索引循环代码快 10 倍。 (顺便说一下,python map 函数是作为解释索引循环实现的——它与 numpy 数组操作无关。)

from numpy import asfarray, broadcast_arrays, arange

# Begin by defining the function to be inverted by Newton's method.
def f_dfdx(x):
    x = asfarray(x)  # always avoid repeated type conversions
    return x**2, 2.*x

# First, the simplest algorithm to find x such that f(x)=y.
# We must supply a starting guess x0 for x.
def f_inverse0(f_dfdx, y, x0, tol=1.e-7):
    y, x = broadcast_arrays(asfarray(y), asfarray(x0))
    x = x.copy()  # without this may clobber input x0
    for npass in range(20):
        f, dfdx = f_dfdx(x)
        dx = (f - y) / dfdx
        if (abs(dx) <= tol).all():
            break  # iterate all x until all have converged
        x -= dx
    else:
        raise RuntimeError("failed to converge")
    return x

# A frequently slower algorithm that avoids extra iterations.
def f_inverse1(f_dfdx, y, x0, tol=1.e-7):
    y, x = broadcast_arrays(asfarray(y), asfarray(x0))
    shape = x.shape
    y, x = y.ravel(), x.flatten()  # avoid clobbering x0
    unconverged = arange(y.size)
    for npass in range(20):
        f, dfdx = f_dfdx(x[unconverged])
        dx = (f - y[unconverged]) / dfdx
        unc = abs(dx) > tol
        unconverged = unconverged[unc]
        if not unconverged.size:
            break  # iterate all x until all have converged
        x[unconverged] -= dx[unc]
    else:
        raise RuntimeError("failed to converge")
    return x.reshape(shape)

在我的机器上,OP 的 C++ 程序运行时间为 2.03 秒(1.64+0.38 用户+系统)。对于 C++ 程序的 n=1 亿,f_inverse0 运行时间为 20.4 秒(4.7+15.6 用户+系统)。正如预期的那样,f_inverse1 较慢,为 51.3 秒(11.5+39.8 用户+系统)。同样,在编写 numpy 代码时,不要自动尝试最小化总操作数。高系统开销可能是由于繁重的内存管理 - 每个临时向量都是 0.8 GB 并且内存管理器正在苦苦挣扎。

将数组大小减少到 n = 100 万个元素 (8 MB),然后将运行时间乘以 100 会大大降低系统时间,f_inverse0 现在需要 16.1 s (12.5+3.6),而 f_inverse1 需要 22.3 s (16.2+5.1)。这个比编译代码慢 8 到 10 倍的因素对于 numpy 性能而言并非不合理。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2018-02-26
    • 2018-05-24
    • 2018-05-19
    • 1970-01-01
    • 1970-01-01
    • 2018-04-09
    • 2018-11-21
    相关资源
    最近更新 更多