标签: parallel-processing

我正在查询 Freebase 以获取大约 10000 部电影的类型信息。

在阅读如何使用 R 中的 getURL() 优化抓取后，我尝试并行执行请求。但是，我失败了 - 见下文。除了并行化，我还读到这httr可能是RCurl.

我的问题是：是否可以通过使用以下循环的并行版本（使用 WINDOWS 机器）来加速 API 调用？b) getURL 的替代方法，例如GET在httr-package 中？

library(RCurl)
library(jsonlite)
library(foreach)
library(doSNOW)

df <- data.frame(film=c("Terminator", "Die Hard", "Philadelphia", "A Perfect World", "The Parade", "ParaNorman", "Passengers", "Pink Cadillac", "Pleasantville", "Police Academy", "The Polar Express", "Platoon"), genre=NA)

f_query_freebase <- function(film.title){

  request <- paste0("https://www.googleapis.com/freebase/v1/search?",
                    "filter=", paste0("(all alias{full}:", "\"", film.title, "\"", " type:\"/film/film\")"),
                    "&indent=TRUE",
                    "&limit=1",
                    "&output=(/film/film/genre)")

  temp <- getURL(URLencode(request), ssl.verifypeer = FALSE) …

我正在使用Parallel.ForEach它，它极大地提高了我的代码的性能，但我对DbContext多线程感到好奇。我知道它不是线程安全的，所以我在需要的地方使用锁。

循环遍历字典并计算统计信息：

Dictionary<string, List<decimal>> decimalStats = new Dictionary<string, List<decimal>>(); // this gets populated in another irrelevant loop

List<ComparativeStatistic> comparativeStats = db.ComparativeStatistics.ToList();
var statLock = new object();

Parallel.ForEach(decimalStats, entry =>
{
    List<decimal> vals = ((List<decimal>)entry.Value).ToList();

    if (vals.Count > 0)
    {
        string[] ids = entry.Key.Split('#');
        int questionId = int.Parse(ids[0]);
        int yearId = int.Parse(ids[1]);
        int adjacentYearId = int.Parse(ids[2]);

        var stat = comparativeStats.Where(l => l.QuestionID == questionId && l.YearID == yearId && l.AdjacentYearID == adjacentYearId).FirstOrDefault();

        if (stat == null)
        { …

    #include<omp.h>
    #include<stdio.h>
    #include<stdlib.h>

    void main(int argc, int *argv[]){


   #pragma omp parallel num_threads(3)
   {

    int tid = omp_get_thread_num();
    printf("Hello world from thread = %d \n",tid);
    if(tid == 0){
        int nthreads = omp_get_num_threads();
        printf("Number of threads = %d\n",nthreads);
    }
   }

  }

我正在学习 OpenMP，但我不明白为什么当我指定了线程数 3 时它只执行一个线程？程序输出：

   Hello world from thread = 0
   Number of threads = 1

我正在从 MATLAB 迁移到 Python，主要是因为 Python 中有大量有趣的机器学习包可用。但让我感到困惑的问题之一是并行处理。特别是，我想从磁盘中for循环读取数千个文本文件，并且我想并行执行。在 MATLAB 中，使用parfor而不是for可以解决问题，但到目前为止我还没有弄清楚如何在 python 中做到这一点。这是我想要做的一个例子。我想读取 N 个文本文件，将它们组成一个 N1xN2 数组，并将每个文件保存到一个 NxN1xN2 numpy 数组中。这个数组将是我从函数返回的内容。假设文件名是file0001.dat，file0002.dat等，我喜欢并行化的代码如下：

import numpy as np
N=10000
N1=200
N2=100
result = np.empty([N, N1, N2])
for counter in range(N):
    t_str="%.4d" % counter        
    filename = 'file_'+t_str+'.dat'
    temp_array = np.loadtxt(filename)
    temp_array.shape=[N1,N2]
    result[counter,:,:]=temp_array

我在集群上运行代码，所以我可以使用许多处理器来完成这项工作。因此，任何关于哪种并行化方法更适合我的任务（如果有多个）的评论都是最受欢迎的。

注意：我知道这篇文章，但在那篇文章中，只有out1, out2,out3变量需要担心，并且它们已被明确用作要并行化的函数的参数。但是在这里，我有许多 2D 数组应该从文件中读取并保存到 3D 数组中。所以，这个问题的答案对我的情况来说不够通用（或者我是这样理解的）。

我们是否有并行版本的 sapply，因为我们在并行包中有 mclapply，它是 lapply 的一个版本。

我有一个我试图并行化的算法，因为串行运行时间很长。然而，需要并行化的函数在一个类中。multiprocessing.Pool似乎是最好和最快的方式来做到这一点，但有一个问题。它的目标函数不能是对象实例的函数。意思是这个；您可以Pool通过以下方式声明 a ：

import multiprocessing as mp
cpus = mp.cpu_count()
poolCount = cpus*2
pool = mp.Pool(processes = poolCount, maxtasksperchild = 2)

然后实际使用它：

pool.map(self.TargetFunction, args)

但这会引发错误，因为无法对对象实例进行腌制，因为该Pool函数确实会将信息传递给其所有子进程。但我必须使用self.TargetFunction

所以我有一个想法，我将创建一个名为的新 Python 文件，parallel并简单地编写几个函数而不将它们放在一个类中，然后从我的原始类（我想并行化其函数）中调用这些函数

所以我试过这个：

import multiprocessing as mp

def MatrixHelper(args):
    WM = args[0][0]
    print(WM.CreateMatrixMp(*args))
    return WM.CreateMatrixMp(*args)

def Start(sigmaI, sigmaX, numPixels, WM):

    cpus = mp.cpu_count()
    poolCount = cpus * 2
    args = [(WM, sigmaI, sigmaX, i) for i in range(numPixels)]
    print('Number of cpu\'s …

我有一些代码可以将 Scrapy 抓取的数据写入 SQL 服务器数据库。数据项包括一些基本的酒店数据（名称、地址、评级...）和一些带有相关数据（价格、入住率等）的房间列表。可以有多个 celery 线程和多个服务器运行此代码并同时写入数据库不同的项目。我遇到死锁错误，例如：

[Failure instance: Traceback: <class 'pyodbc.ProgrammingError'>: 
('42000', '[42000] [FreeTDS][SQL Server]Transaction (Process ID 62)
 was deadlocked on lock resources with another process and has been 
chosen as the deadlock victim. Rerun the transaction. (1205) (SQLParamData)')

实际执行插入/更新的代码示意如下：

1) Check if hotel exists in hotels table, if it does update it, else insert it new. 
   Get the hotel id either way. This is done by `curs.execute(...)`

2) Python loop over the hotel rooms scraped. For each room …

问题

有几个任务是并行运行的,all,none或者其中任何一个都可能抛出异常.当所有任务完成后,必须报告所有可能发生的异常(通过日志,电子邮件,控制台输出......等等).

预期的行为

我可以通过linq和async lambdas构建所有任务,然后等待它们并行运行Task.WhenAll(tasks).然后我可以抓住AggregateException并报告每个内部异常.

实际行为

一个AggregateException被抛出,但它仅包含一个内部异常,无论个别例外的数量已经抛出.

最小的完整可验证的例子

static void Main(string[] args)
{
    try
    {
        ThrowSeveralExceptionsAsync(5).Wait();
    }
    catch (AggregateException ex)
    {
        ex.Handle(innerEx =>
        {
            Console.WriteLine($"\"{innerEx.Message}\" was thrown");
            return true;
        });
    }

    Console.ReadLine();
}

private static async Task ThrowSeveralExceptionsAsync(int nExceptions)
{
    var tasks = Enumerable.Range(0, nExceptions)
        .Select(async n =>
        {
            await ThrowAsync(new Exception($"Exception #{n}"));
        });

    await Task.WhenAll(tasks);
}

private static async Task ThrowAsync(Exception ex)
{
    await Task.Run(() => {
        Console.WriteLine($"I am going to throw \"{ex.Message}\""); …

我正在改进一个解析XML并对其子树进行分类和索引的程序.实际程序太大而无法显示在这里,所以我把它归结为一个最小的测试用例,显示我遇到的问题.

这个想法是:

1. 逐个处理目录中的XML文件
2. alpino_ds并行处理文件中的所有节点
3. 在此过程中,该进程需要对共享变量进行读/写访问,以便我们可以检查属性总共发生了多少次,或者跟踪文件句柄

请注意,在实际代码中还有一些注意事项:

• simply returning new values per process and then merging them in the main thread seems not advisable and presumably quite slow because the actual data structure are dicts of four levels deep consisting of dicts, sets, ints, and strings, as well as dict-to-filehandle, and Counter() objects;
• I tried using threads (with ThreadPoolExecutor) and even though there was some gain (I calculated around 5% improvement in speed), this …

多处理模块对于python初学者来说非常困惑,特别是那些刚刚从MATLAB迁移并且使用并行计算工具箱变得懒惰的人.我有以下功能需要大约80秒运行,我想通过使用Python的多处理模块来缩短这个时间.

from time import time

xmax   = 100000000

start = time()
for x in range(xmax):
    y = ((x+5)**2+x-40)
    if y <= 0xf+1:
        print('Condition met at: ', y, x)
end  = time()
tt   = end-start #total time
print('Each iteration took: ', tt/xmax)
print('Total time:          ', tt)

这按预期输出:

Condition met at:  -15 0
Condition met at:  -3 1
Condition met at:  11 2
Each iteration took:  8.667453265190124e-07
Total time:           86.67453265190125

由于循环的任何迭代都不依赖于其他循环,我尝试从官方文档中采用此服务器进程来在单独的进程中扫描范围的块.最后我想出了vartec对这个问题的回答,可以准备以下代码.我还根据Darkonaut对当前问题的回答更新了代码.

from time import time 
import multiprocessing as …