标签: parallel-processing

有没有办法阻止我的循环被速率限制中断？如果可能的话，我希望我的代码等到时间限制过后才执行。

一个附带问题：我考虑过并行化 for 循环。你认为这会是个好主意吗？我不确定是否有机会将数据写入错误的文件。

library(rtweet)
create_token(app="Arconic Influential Followers",consumer_key,consumer_secret) 

flw <- get_followers("arconic")
fds <- get_friends("arconic")
usrs <- lookup_users(c(flw$user_id, fds$user_id))

for(i in 1:length(usrs$user_id)){

    a<-tryCatch({get_timeline(usrs$user_id[i])},
                error=function(e){message(e)}
       )
    tryCatch({save_as_csv(a,usrs$user_id[i])},
                error=function(e){message(e)}
       )

}

假设我有一个 8 核 CPU。doParallel在 R 中使用，当我注册时makeCluster(x)，理想的核心数是x多少？

是尽可能多的核心吗？或者使用 7 核会比使用 6 核慢吗？有什么规则可以解决这个问题吗？

我试图了解 Spark 并行性中 Java 8 并行流的行为。当我运行下面的代码时，我期望的输出大小listOfThings与输入大小相同。但事实并非如此，我有时会在输出中丢失项目。这种行为是不一致的。如果我只是遍历迭代器而不是使用parallelStream，那么一切都很好。每次都计算匹配。

// listRDD.count = 10
JavaRDD test = listRDD.mapPartitions(iterator -> {
    List listOfThings = IteratorUtils.toList(iterator);
    return listOfThings.parallelStream.map(
        //some stuff here
    ).collect(Collectors.toList());
});
// test.count = 9
// test.count = 10
// test.count = 8
// test.count = 7

我正在尝试通过hargreaves方法计算蒸发量package SPEI。这涉及使用最低温度 ( TMIN) 和最高温度 ( TMAX)。鉴于此Tmin，并行计算是我最好的选择，并且Tmax rasterstacks拥有500,000 cells and 100 layers each. Hargreaves function取Tmin，Tmax并latitude在each grid作为输入。以下是我的第一个猜测如何解决这个问题：

library(SPEI)
# go parallel 
library(parallel)
clust <- makeCluster(detectCores())

#har <- hargreaves(TMIN,TMAX,lat=37.6475) # get evaporation for a station. 

但是，我的数据是网格化的。

Tmin并且Tmax是名单中，每个数据帧Tmin，并Tmax有一个$latitude连接到它。在 中pet，k$d是 Tmin，k$d是 Tmax（也许我应该在petegfunction(k,y)而不是仅仅提供两个参数k？） …

该包的摘要的java.util.stream规定如下：

有状态 lambda 的一个例子是map()in的参数：

Set<Integer> seen = Collections.synchronizedSet(new HashSet<>());
stream.parallel().map(e -> { if (seen.add(e)) return 0; else return e; })...

Run Code Online (Sandbox Code Playgroud)

在这里，如果映射操作是并行执行的，由于线程调度差异，相同输入的结果可能会因运行而异，而对于无状态 lambda 表达式，结果将始终相同。

我不明白为什么这不会产生一致的结果，因为该集合是同步的并且一次只能处理一个元素。您能否以一种演示结果如何因并行化而变化的方式完成上述示例？

并行 STL 算法是否符合std::back_insert_iterator??

我可能误解了std::par和之间的区别std::par_vec，是否std::par_vec意味着需要预先分配输出范围？

代码示例：

auto numbers = {1,2,3,4,5,6};
auto squared = std::vector<int>{};
std::transform(
  **std::par/std::par_vec,**
  numbers.begin(),
  numbers.end(),
  std::back_inserter(squared),
  [](auto val) { 
    return val*val; 
  }
);

更新

简化问题作为我的第一个问题是误读文章的结果。

I have an IEnumerable<IEnumerable<T>> method called Batch that works like

var list = new List<int>() { 1, 2, 4, 8, 10, -4, 3 }; 
var batches = list.Batch(2); 
foreach(var batch in batches)
    Console.WriteLine(string.Join(",", batch));

-->

1,2
4,8
10,-4
3

The problem I've having is that I'm to optimize something like

foreach(var batch in batches)
    ExecuteBatch(batch);

by

Task[] tasks = batches.Select(batch => Task.Factory.StartNew(() => ExecuteBatch(batch))).ToArray();
Task.WaitAll(tasks);

or

Action[] executions = batches.Select(batch => new Action(() => ExecuteBatch(batch))).ToArray();
var options = new ParallelOptions …

我可以使用具有多个 CPU 内核（即 56 个）的计算机，并且在使用 Tensorflow 训练模型时，我希望通过使每个内核成为模型的独立训练器来最大限度地利用上述内核。

在 Tensorflow 的文档中，我发现这两个参数（Inter 和 Intra Op 并行度）在训练模型时控制并行度。但是，这两个参数不允许执行我的意图。

我怎样才能让每个核心成为独立的工人？（即，一批样本由每个worker分片，然后每个worker根据分配的样本计算梯度。最后，每个worker根据它的梯度更新变量（由所有worker共享）计算过。

我正在测试该类cv::ParallelLoopBody的图像处理代码。

我首先开始实现归一化，在那里我必须为每个通道划分具有特定值的所有像素，这是一个简单的并行代码。

但是，在测试它时，我没有看到任何区别。

我在这里做错了吗？

这是我的课：

class Parallel_process : public cv::ParallelLoopBody
    {

    private:
        cv::Mat img; //my image to normalize
        std::vector<int> A;
        int diff;

    public:
        Parallel_process(cv::Mat inputImage, std::vector<int> AA, int diffVal)
                           : img(inputImage), A(AA), diff(diffVal){}

        virtual void operator()(const cv::Range& range) const
        {
            for(int i = range.start; i < range.end; i++)
            {
              //in is a patch of my original image
               cv::Mat in(img, cv::Rect(0, (img.rows/diff)*i, img.cols, img.rows/diff));
               std::vector<int> AAA (A);
               in.forEach<cv::Vec3f>
                (
                  [&AAA](cv::Vec3f &pixel, const int* po) -> void
                  {
                    pixel[0]/=AAA[0];
                    pixel[1]/=AAA[1]; …

我有两个函数func1()，func2()它们相互独立，可以在两个线程中运行。我正在使用threading库python3来运行这两个线程。不过里面func1()，我也运行for使用并行循环joblib。因此，使用线程以及 joblib 会给出以下警告 -

/usr/local/lib/python3.6/dist-packages/joblib/parallel.py:547: UserWarning: Multiprocessing-backed parallel loops cannot be nested below threads, setting n_jobs=1,

然后里面的 for 循环func1()只是按顺序运行。

以下是代码片段的示例：

from joblib import Parallel, delayed
import threading
from math import sqrt

class MyThread(threading.Thread):
    def __init__(self, sample, type):
        threading.Thread.__init__(self)
        self.type = type
        self.sample = sample
    def run(self):
        if self.type=='func1':
            self.sample.func1()
        else:
            self.sample.func2()

class Sample:
   def func1(self):
       print('this function runs for …