如何使用多处理来解决令人尴尬的并行问题？

令人尴尬的并行问题通常包括三个基本部分:

1. 读取输入数据(来自文件,数据库,tcp连接等).
2. 对输入数据运行计算,其中每个计算独立于任何其他计算.
3. 写入计算结果(到文件,数据库,tcp连接等).

我们可以在两个方面并行化程序:

• 第2部分可以在多个核上运行,因为每个计算都是独立的; 处理顺序无关紧要.
• 每个部分都可以独立运行.第1部分可以将数据放在输入队列中,第2部分可以从输入队列中提取数据并将结果放到输出队列中,第3部分可以将结果从输出队列中拉出并写出来.

这似乎是并发编程中最基本的模式,但我仍然试图解决它,所以让我们写一个规范的例子来说明如何使用多处理来完成.

下面是示例问题:给定一个包含整数行作为输入的CSV文件,计算它们的总和.将问题分成三个部分,这些部分可以并行运行:

1. 将输入文件处理为原始数据(整数的列表/可迭代)
2. 并行计算数据的总和
3. 输出总和

下面是传统的单进程绑定Python程序,它解决了以下三个任务:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# basicsums.py
"""A program that reads integer values from a CSV file and writes out their
sums to another CSV file.
"""

import csv
import optparse
import sys

def make_cli_parser():
    """Make the command line interface parser."""
    usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV",
            __doc__,
            """ …

我有一个产生数字的函数(神经网络模型).我希望在带有Torque的标准集群上使用PBS从python测试几个参数,方法和不同输入(意味着数百次运行的函数).

注意:我尝试了parallelpython,ipython等,并且从未完全满意,因为我想要更简单的东西.集群处于给定的配置中,我无法改变,这样的集成python + qsub的解决方案肯定会对社区有益.

为简化起见,我有一个简单的功能,例如:

import myModule
def model(input, a= 1., N=100):
    do_lots_number_crunching(input, a,N)
    pylab.savefig('figure_' + input.name + '_' + str(a) + '_' + str(N) + '.png')

input表示输入的对象在哪里,input.name是一个字符串,do_lots_number_crunching可能持续数小时.

我的问题是:是否有正确的方法来转换像扫描参数之类的东西

for a in pylab.linspace(0., 1., 100):
    model(input, a)

进入"东西",为每次调用model函数启动PBS脚本？

#PBS -l ncpus=1
#PBS -l mem=i1000mb
#PBS -l cput=24:00:00
#PBS -V
cd /data/work/
python experiment_model.py

我正在考虑一个包含PBS模板的函数,并从python脚本中调用它,但还不能解决它(装饰器？).

我正在寻找能够在群集上轻松运行(正确编码)令人尴尬的并行JVM代码的东西(这样我就可以使用Clojure + Incanter).

我过去曾使用Parallel Python来做到这一点.我们有一个新的PBS集群,我们的管理员很快将设置使用PBS作为后端的IPython节点.这两个系统都使得在集群中运行某些类型的代码几乎是明智之举.

我错误地在过去使用Hadoop(Hadoop不适合我使用的那种数据) - 甚至小的运行延迟执行1-2分钟.

JPPF或Gridgain更适合我需要的吗？这里有没有人有任何经验？还有什么可以推荐的吗？

我想对从文件加载的一些输入数据运行并行计算.(文件可能非常大,所以我使用了一个生成器.)

在一定数量的项目上,我的代码运行正常但超过此阈值程序挂起(某些工作进程不会结束).

有什么建议？(我用python2.7运行它,8个CPU; 5000行仍然可以,7,500不起作用.)

首先,您需要一个输入文件.在bash中生成它:

for i in {0..10000}; do echo -e "$i"'\r' >> counter.txt; done

然后,运行这个:

python2.7 main.py 100 counter.txt > run_log.txt

main.py:

#!/usr/bin/python2.7
import os, sys, signal, time
import Queue
import multiprocessing as mp

def eat_queue(job_queue, result_queue):
    """Eats input queue, feeds output queue
    """
    proc_name = mp.current_process().name
    while True:
        try:
            job = job_queue.get(block=False)
            if job == None:
                print(proc_name + " DONE")
                return
            result_queue.put(execute(job))
        except Queue.Empty:
            pass    

def execute(x):
    """Does the computation on the input data
    """
    return x*x …

我们遇到一个令人尴尬的并行问题 - 我们运行一个单个程序的大量实例,每个程序都有不同的数据集; 我们只需将应用程序多次提交到具有不同参数的批处理队列即可.

但是,由于有大量工作,并非所有工作都完成.它似乎不是队列中的问题 - 所有作业都已启动.

问题似乎是,随着应用程序的大量实例运行,许多作业大致同时完成,因此所有人都试图在几乎同时将其数据写入并行文件系统.

然后问题似乎是程序无法写入文件系统并以某种方式崩溃,或者只是坐在那里等待写入,批处理队列系统在等待太长时间后就会终止作业.(从我收集的问题来看,大多数未能完成的工作,如果不是全部,都不会留下核心文件)

调度磁盘写入以避免此问题的最佳方法是什么？我提到我们的程序是令人尴尬的并行突出显示每个进程都不知道其他进程的事实 - 他们不能互相交谈以某种方式安排他们的写入.

虽然我有程序的源代码,但我们希望解决问题而不必在可能的情况下修改它,因为我们不维护或开发它(加上大多数注释都是意大利语).

我对此事有一些想法:

1. 每个作业首先写入节点的本地(暂存)磁盘.然后我们可以运行另一个工作,它会立即检查已完成的作业,并将文件从本地磁盘移动到并行文件系统.
2. 在主/从系统中使用围绕程序的MPI包装器,其中主服务器管理作业队列并将这些作业关闭到每个从属服务器; 从属包装器运行应用程序并捕获异常(我可以为C++中的文件系统超时可靠地执行此操作,或者可能是Java吗？),并将消息发送回主服务器以重新运行作业

与此同时,我需要纠缠我的主管以获取有关错误本身的更多信息 - 我从未亲自遇到过它,但我还没有将程序用于大量数据集(尚未).

如果它有用:我们在HPC系统上使用SGE(Sun GridEngine)批处理队列系统运行Solaris.文件系统是NFS4,存储服务器也运行Solaris.HPC节点和存储服务器通过光纤通道链路进行通信.

有谁知道GCE是否有Starcluster等价物？我非常高兴使用Starcluster和EC2进行令人尴尬的并行工作.现在我想试试GCE.我很乐意为可能正在进行的任何项目做出贡献,但我没有找到任何东西.

现在,我想我只需手动安装我需要的软件(即Open Grid Engine)并生成一个复制图像.不是太难,但我想我先在这里查看.

在扩展使用云实例上的所有内核时,有没有其他人注意到可怕的性能有些内存紧张的工作(在我的情况下为2.5GB)？

当我在我的四核至强芯片上本地运行作业时,使用1核和所有4核之间的差异是所有核心减慢约25%.这是我所理解的; 随着内核耗尽,时钟速率下降是多核芯片设计的一部分.

但是,当我在多核虚拟实例上运行作业时,我看到使用1核和所有核之间的处理时间减少了2x - 4x.我在GCE,EC2和Rackspace实例上看过这个.我测试了许多不同的实例类型,大多数是最快的.

因此,其他人在内存使用量相同的情况下会看到这种行为吗？

我正在运行的工作是用fortran编写的.我没有写它们,我不是一个真正的强人,所以我对它们的了解是有限的.我知道他们的I/O需求很低.当我在跑步时看顶部时,它们似乎是CPU限制的.它们在不需要彼此通信的情况下运行,即,令人尴尬的并行.它们每个内存大约需要2.5GB.

所以到目前为止我最好的猜测是,占用这么多内存的工作会受到虚拟化层内存管理的巨大冲击.也可能是我的工作正在竞争I/O资源,但据专家称,这似乎不太可能.

我现在的解决方法是使用GCE,因为他们拥有的单核实例实际上可以像笔记本电脑的芯片那样快速地运行作业,并且几乎按核心价格成比例.

我需要在Google Compute Engine上设置并运行"令人尴尬"的并行作业.我正在寻找工具来促进这一点.

在EC2上,我使用MIT的Starcluster来设置集群,然后将作业提交给SGE.

Google Compute Engine还有类似的工具吗？

如果没有,我可能会手动设置一个Condor Cluster来完成这些工作.嗡嗡杀人.

我正在学习使用Python的Multiprocessing软件包来解决令人尴尬的并行问题,因此我编写了串行和并行版本来确定小于或等于自然数n的素数的数量.根据我从博客文章和Stack Overflow问题中读到的内容,我想出了以下代码:

串行

import math
import time

def is_prime(start, end):
    """determine how many primes within given range"""
    numPrime = 0
    for n in range(start, end+1):
        isPrime = True
        for i in range(2, math.floor(math.sqrt(n))+1):
            if n % i == 0:
                isPrime = False
                break
        if isPrime:
            numPrime += 1
    if start == 1:
        numPrime -= 1  # since 1 is not prime
    return numPrime

if __name__ == "__main__":
    natNum = 0
    while natNum < 2: …

熊猫的新手，我已经想要并行执行逐行应用操作。到目前为止，我发现Parallelize在pandas groupby之后适用。但是，这似乎仅适用于分组的数据帧。

我的用例是不同的：我有一个假期列表，并且对于我当前的行/日期，想要找到从这一天之前到第二天到下一个假期的无休日。

这是我通过apply调用的函数：

def get_nearest_holiday(x, pivot):
    nearestHoliday = min(x, key=lambda x: abs(x- pivot))
    difference = abs(nearesHoliday - pivot)
    return difference / np.timedelta64(1, 'D')

我如何加快速度？

编辑

我对pythons池做了一些实验-但这既不是很好的代码，也没有得到我的计算结果。

假设我只有一个函数processing。我想为多个参数并行运行相同的函数多次，而不是一个接一个地依次运行。

def processing(image_location):
    
    image = rasterio.open(image_location)
    ...
    ...
    return(result)

#calling function serially one after the other with different parameters and saving the results to a variable.
results1 = processing(r'/home/test/image_1.tif')
results2 = processing(r'/home/test/image_2.tif')
results3 = processing(r'/home/test/image_3.tif')

例如，如果我运行delineation(r'/home/test/image_1.tif')然后delineation(r'/home/test/image_2.tif')，然后delineation(r'/home/test/image_3.tif')，如图上面的代码，这将顺序运行一前一后，并且如果需要5分钟一个函数来运行然后运行这三个将采取5X3 = 15分钟。因此，我想知道我是否可以并行/尴尬地并行运行这三个，以便对所有三个不同参数执行该函数只需要 5 分钟。

帮助我以最快的方式完成这项工作。该脚本应该能够利用默认情况下可用的所有资源/CPU/ram 来完成此任务。