Dask：凸矩阵优化

Posted 2023-03-12

【中文标题】Dask：凸矩阵优化

【英文标题】：Dask: Convex Matrix Optimization

【发布时间】：2020-03-10 16:57:35

【问题描述】：

我目前正在尝试实现 Dask 以进行凸矩阵优化。目标是对内存不足的矩阵执行矩阵优化（分解）。使用一个高矩阵作为输入，两个高矩阵作为输出，以及一些参数（例如收敛等），我使用 dask 数组来分块我的原始矩阵和迭代/输出矩阵。最后，迭代是连续的，因为前一次迭代的输出被用作新迭代的输入（参见下面的简单示例）。

在每次迭代中，必须计算并检查两个标准是否收敛（最终的 if 语句）。但是，如果我执行下面给出的代码，我看到的是 Dask 计算标准（由 if 语句强制执行），然后在每次迭代中重新计算其他矩阵：即迭代 1 标准计算正确，迭代2 没有使用之前找到的 A 和 E，而是重新计算它们，导致两次 SVD 评估等等。 Dask 似乎没有意识到 A_hat 和 E_hat 已经在之前的迭代中计算过了，然后从一开始就重新计算了这些值。反之亦然，它会根据标准重新计算所有步骤。

将这些矩阵保留在内存中也不是一种选择，因为它们会随着时间的推移在内存中建立。我在博文（https://matthewrocklin.com/blog/work/2017/03/22/dask-glm-1 和http://matthewrocklin.com/blog/work/2017/04/19/dask-glm-2）中看到了类似的优化，但似乎每次迭代都有可能在内存中获得结果，而对于我的优化来说这是不可能的。我尝试了不同的设置，在这些设置中我持久化了一些值，或者使用了同步和异步持久化和计算。

因此，我的问题是，如何在每次迭代结束时计算矩阵（A 和 E，以及扩展为 Y）而不被计算并放置在内存中，同时在每次迭代结束时计算它们的一些规范？我有一些使用期货的想法，或者明确延迟所有单独的步骤。其中一个可以解决我的问题还是有更有效的方法？

提前致谢，

罗杰

def PCP(D, lambda_i, kmax):
    
    # Initialize norms
    normD = (norm(D, 'fro')).compute()
    normD2 = (norm(D, 2)).compute()
    normDinf = (norm(D.ravel(), np.inf)).compute()

    # Initialize Y
    Y = (1/np.maximum(normD2, (1/lambda_i)*normDinf))*D

    # Initialize A and A_hat
    A = da.zeros_like(D)
    A_hat = da.zeros_like(D)
    
    # Initialize E and E_hat
    E = da.zeros_like(D)
    E_hat = da.zeros_like(D)

    # Initialize mu: mu_0 > 0
    mu = 1.25/normD2 # Convergence Rate

    # Initialize rho: rho > 1
    rho = 1.6 # Convergence Rate

    runAlgorithm = True
    k = 0
    while runAlgorithm:
        
        # Estimate E_hat
        E_hat = da.maximum((D-A+(1/mu)*Y)-(1/mu)*lambda_i, 0)+da.minimum((D-A+(1/mu)*Y)+(1/mu)*lambda_i, 0)
        
        # Estimate A_hat
        Q = da.linalg.svd(D-E_hat+(1/mu)*Y)
        svp = (Q[1] - 1/mu).clip(0)
        A_hat = (Q[0]*svp)@Q[2]
        
        # Update Y
        Y = Y+mu*(D-A_hat-E_hat)
        
        # Calculate stopping criteria
        crit1 = norm(D-A_hat-E_hat, 'fro')/normD
        crit2 = norm(E-E_hat, 'fro')/normD
        
        # Update A and E
        A = A_hat
        E = E_hat
        
        # Check if Converged
        if k == kmax or crit1 < 0.01 or crit2 < 0.01:
            runAlgorithm = False
    
        k = k + 1
        

    return  A, E, k, crit1, crit2

我使用以下 dask 分布式设置：

from dask.distributed import Client, progress
client = Client(threads_per_worker=4,
            n_workers=1, memory_limit='4GB')

【参考方案1】：

我不确定我是否完全理解您的问题，但我会在这里说一些您可能已经知道的一般性内容。

如果您可以将某些结果放入内存中，那么您应该考虑在这些部分上调用persist 以避免重复工作。如果您无法将它们放入内存中，那么我不确定在这里可以做什么。像你上面描述的那样制作高效的算法很大程度上是为了弄清楚你需要将计算的哪些部分保持为小状态，以及每次都需要重新计算哪些部分。

例如，您可能想看看matrix algorithms in dask-glm，它有效地表演了这种舞蹈。