标签: parallel-processing

我正在使用Stream并行处理,并了解如果我使用平面阵列流,它会得到非常快速的处理.但如果我使用ArrayList,那么处理速度会慢一些.但是,如果我使用LinkedList或使用一些二进制树,处理速度会更慢.

所有听起来更像是流的可分割性,处理速度越快.这意味着阵列和数组列表在并行流的情况下最有效.这是真的吗？如果是这样,ArrayList如果我们想并行处理流,我们总是使用或者Array吗？如果是这样,如何使用LinkedList和BlockingQueue并行流？

另一件事是选择的中间函数的状态.如果我执行像无状态操作filter(),map(),性能高,但如果执行像国家提供充分的操作distinct(),sorted(),limit(),skip(),它需要大量的时间.再次,并行流变慢.这是否意味着我们不应该在并行流中使用状态全中间函数？如果是这样,那么解决这个问题的方法是什么？

我正在使用这样的东西来并行运行测试:

stage('Test') {
  steps {
    script {
      testing_closures = [one: { print("staring one"); sleep 10; print("finishing one") },
                          two: { print("staring two"); sleep 10; print("finishing two") },
                          three: { print("staring three"); sleep 10; print("finishing three") },
                          four: { print("staring four"); sleep 10; print("finishing four") },
                          five: { print("staring five"); sleep 10; print("finishing five") },
                          six: { print("staring six"); sleep 10; print("finishing six") }]
      parallel(testing_closures)
    }
  }
}

主要目标是限制那些关闭 - 我不希望它们中的所有六个同时运行 - 一次只有3个.我希望能够运行另一个这样的构建,它也将运行所有这些闭包,但同时只运行3个.

我正在考虑使用节点 - 即在节点{}块中包装每个闭包:

one: { node { print("staring …

我试图在一个非常大的数据集上运行一些东西.基本上,我想遍历文件夹中的所有文件并在其上运行fromJSON函数.但是,我希望它跳过产生错误的文件.我已经使用tryCatch构建了一个函数,但只有在我使用函数lappy而不是parLapply时才有效.

这是我的异常处理函数的代码:

readJson <- function (file) {
 require(jsonlite)
 dat <- tryCatch(
        {
         fromJSON(file, flatten=TRUE)      
        },
         error = function(cond) {
                 message(cond)
                 return(NA)
        },
         warning = function(cond) {
                  message(cond)
                  return(NULL)
                  }
   )
  return(dat)   
}

然后我在包含JSON文件的完整路径的字符向量文件上调用parLapply :

 dat<- parLapply(cl,files,readJson)

当它到达一个未正确结束的文件时会产生错误,并且不会通过跳过有问题的文件来创建列表'dat'.这是readJson函数应该缓解的内容.

当我使用常规lapply,但它工作得很好.它会生成错误,但是,它仍然会跳过错误的文件来创建列表.

关于如何使用parLappy并行处理异常处理的任何想法,以便它会跳过有问题的文件并生成列表？

在我的Shiny应用程序中,用户可以生成重型powerpoint报告.当它包含大量幻灯片时,可能需要30分钟才能完成.因此,我希望在独立的流程/任务中处理这些任务,即使应用程序关闭也可以工作 - 例如,用户单击按钮生成报告,关闭应用程序,报告准备好时,应用程序通过电子邮件通知用户.是否有任何良好做法或经过验证的解决方案？

我的第一个想法是使用future套装plan(multisession)- 但我不确定当用户关闭应用程序时会发生什么 - future会话关闭或不关闭？

我一直在编写一个脚本,我希望通过交换来mapM利用我机器中的多个处理器Async.mapConcurrently.

在该实例中没有观察到速度增加,我想验证runghc确实可以使用多个核心.

给定一个文件Foo.hs:

import Control.Concurrent

main = print =<< Control.Concurrent.getNumCapabilities

如果我按如下方式编译文件:

stack ghc -- -threaded Foo.hs

然后运行如下:

./Foo

它返回结果1.这是预期的,因为没有提供RTS选项.运行它如下:

./Foo +RTS -N

返回数字6,因为我的机器中有6个处理器(同意nproc).

但是,当我以"解释模式"运行脚本时,如下所示:

GHCRTS="-N" stack runghc Foo.hs

它会产生以下错误文本:

Running /home/kostmo/.stack/programs/x86_64-linux/ghc-nopie-8.0.2/bin/ghc-pkg --no-user-package-db list --global exited with ExitFailure 1

ghc-pkg: the flag -N requires the program to be built with -threaded

是否有可能利用堆栈"脚本"多个核心？

我想知道我是否可以加快排列的产生.具体来说,我正在使用[az]中的8个,我想使用[a-zA-Z]中的8个和[a-zA-Z0-9]中的8个.我所知道的将很快占用大量的时间和空间.

即使仅使用小写ASCII字符的长度为8的排列也需要一段时间并生成千兆字节.我的问题是我不理解底层算法,所以我无法弄清楚我是否可以将问题分解成比我以后可以加入的更小的任务.

我用来生成排列列表的python脚本:

import string
import itertools
from itertools import permutations

comb = itertools.permutations(string.ascii_lowercase, 8)

f = open('8letters.txt', 'w')
for x in comb:
        y = ''.join(x)
        f.write(y + '\n')

f.close()

有谁知道如何将其划分为子任务并将它们放在一起？有可能吗？

我可能只是尝试(可能)更快的方式,但我遇到了C++及其std :: next_permutation()的问题,所以我无法验证它是否可以加速甚至一点点.

如果我可以将它分成16个任务,并在16个Xeon CPU上运行,那么加入结果,这将是很棒的.

我想在for循环中使用shell脚本,并行运行100个文件.

目前,我有一个以下格式的shell脚本:

#!/bin/bash
NUM=10
python a1.py $((NUM + 0)) &
python a2.py $((NUM + 2)) &
python a3.py $((NUM + 4)) &
python a4.py $((NUM + 6)) &
python a5.py $((NUM + 8)) &

现在,如果我有a1.py,a2.py,a3.py... ... a100.py,我想并行运行它们,我怎么做,在for循环？

我正在创建一个DocumentTermMatrix使用create_matrix()来RTextTools创建container并model基于它.它适用于极大的数据集.

我为每个类别(因子级别)执行此操作.因此,对于每个类别,它必须运行矩阵,容器和模型.当我运行下面的代码(例如16核/ 64 GB)时 - 它只在一个核心中运行,并且使用的内存小于10%.

有没有办法加快这个过程？也许用doparallel&foreach？任何信息肯定会有所帮助.

#import the required libraries
library("RTextTools")
library("hash")
library(tm)

for ( n in 1:length(folderaddress)){
    #Initialize the variables
    traindata = list()
    matrix = list()
    container = list()
    models = list()
    trainingdata = list()
    results = list()
    classifiermodeldiv = 0.80`

    #Create the directory to place the models and the output files
    pradd = paste(combinedmodelsaveaddress[n],"SelftestClassifierModels",sep="")
    if (!file.exists(pradd)){
        dir.create(file.path(pradd))
    }  
    Data$CATEGORY <- as.factor(Data$CATEGORY)

    #Read the …

在另一个Q + A(我可以在pandas中执行动态cumsum？)我对使用prange这个代码的正确性做了评论(这个答案):

from numba import njit, prange

@njit
def dynamic_cumsum(seq, index, max_value):
    cumsum = []
    running = 0
    for i in prange(len(seq)):
        if running > max_value:
            cumsum.append([index[i], running])
            running = 0
        running += seq[i] 
    cumsum.append([index[-1], running])

    return cumsum

评论是:

我不建议并行化一个不纯的循环.在这种情况下,running变量使其不纯.有4种可能的结果:(1)numba决定它不能并行处理它只是处理循环cumsum而不是prange(2)它可以将变量提升到循环之外并在余数上使用并行化(3)numba错误地插入并行执行和结果之间的同步可能是虚假的(4)numba在运行时插入必要的同步,这可能会比通过并行化首先获得更多的开销

而后来的补充:

当然,running和cumsum变量都使循环"不纯",而不仅仅是前面评论中所述的运行变量

然后我被问到:

这可能听起来像一个愚蠢的问题,但我怎么能弄清楚它做了哪4件事并改进了呢？我真的想用numba变得更好!

鉴于它可能对未来的读者有用,我决定在这里创建一个自我回答的Q + A. 掠夺者:我无法真正回答4个结果中的哪一个产生的问题(或者如果numba产生完全不同的结果),所以我非常鼓励其他答案.

从我发现的内容来看，还有一个其他问题（加速嵌套交叉验证），但是尝试在此站点和Microsoft上提出了一些修复建议后，安装MPI对我也不起作用，所以我希望有另一个软件包或回答这个问题。

我正在寻找比较多种算法和gridsearch各种参数（也许参数太多？）的方法，除了mpi4py之外还有什么方法可以加快我的代码的运行速度？据我了解，我不能使用n_jobs = -1，因为那是不嵌套的？

还要注意，我无法在下面尝试查看的许多参数上运行它（运行时间超过了我的时间）。如果我给每个模型仅两个参数进行比较，则只有2小时后才会有结果。另外，我在252行和25个特征列以及4个类别变量的数据集上运行此代码，以预测（“确定”，“可能”，“可能”或“未知”）某个基因（具有252个基因）是否影响疾病。使用SMOTE将样本大小增加到420，然后将其投入使用。

dataset= pd.read_csv('data.csv')
data = dataset.drop(["gene"],1)
df = data.iloc[:,0:24]
df = df.fillna(0)
X = MinMaxScaler().fit_transform(df)

le = preprocessing.LabelEncoder()
encoded_value = le.fit_transform(["certain", "likely", "possible", "unlikely"])
Y = le.fit_transform(data["category"])

sm = SMOTE(random_state=100)
X_res, y_res = sm.fit_resample(X, Y)

seed = 7
logreg = LogisticRegression(penalty='l1', solver='liblinear',multi_class='auto')
LR_par= {'penalty':['l1'], 'C': [0.5, 1, 5, 10], 'max_iter':[500, 1000, 5000]}

rfc =RandomForestClassifier()
param_grid = {'bootstrap': [True, False],
              'max_depth': [10, 20, 30, 40, 50, 60, 70, 80, 90, 100, None],
              'max_features': ['auto', 'sqrt'], …