标签: parallel-processing

我对 Spark 很陌生，我试图在网上寻找一些东西，但我没有找到任何令人满意的东西。

我一直使用该命令运行并行计算mclapply，我喜欢它的结构（即，第一个参数用作滚动索引，第二个参数是要并行化的函数，然后是传递给函数的其他可选参数）。现在我试图通过 Spark 做同样的事情，即，我想在 Spark 集群的所有节点之间分配我的计算。这就是我所学到的以及我认为应该如何构建代码的内容（我正在使用包sparklyr）：

1. 我使用命令创建到 Spark 的连接spark_connect；
2. 我在 Spark 环境中复制我的 data.framecopy_to并通过它的tibble访问它；
3. 我想实现的“星火友好”的版本mclapply，但我已经看到有在包（我已经看到了存在的功能没有类似功能spark.lapply的SparkR包，但不幸的是它是不是在CRAN了）。

下面是我实现的一个简单的测试脚本，它使用函数mclapply.

#### Standard code that works with mclapply #########
dfTest = data.frame(X = rep(1, 10000), Y = rep(2, 10000))

.testFunc = function(X = 1, df, str) {
    rowSelected = df[X, ]
    y = as.numeric(rowSelected[1] + rowSelected[2])
    return(list(y = y, str = str))
}

lOutput = mclapply(X = 1 …

可能看起来像一个微不足道的问题，但不幸的是无法让它工作

这是代码

from("direct:START")
.process( (ex) -> {
   List<Integer> pages = IntStream.range(1,5).boxed().collect(Collectors.toList());
   ex.getOut().setBody( pages );
 })
.split(body())
.parallelProcessing()
.to("http://someurl?page=${body}");
 --> Get the collective body here

如何获得这份工作！

我试图理解为什么std::for_each在单线程上运行的~3速度比__gnu_parallel::for_each下面的例子快几倍：

Time =0.478101 milliseconds

对比

Time =0.166421 milliseconds

这是我用来进行基准测试的代码：

#include <iostream>
#include <chrono>
#include <parallel/algorithm>

//The struct I'm using for timming
struct   TimerAvrg
{
    std::vector<double> times;
    size_t curr=0,n;
    std::chrono::high_resolution_clock::time_point begin,end;
    TimerAvrg(int _n=30)
    {
        n=_n;
        times.reserve(n);
    }

    inline void start()
    {
        begin= std::chrono::high_resolution_clock::now();
    }

    inline void stop()
    {
        end= std::chrono::high_resolution_clock::now();
        double duration=double(std::chrono::duration_cast<std::chrono::microseconds>(end-begin).count())*1e-6;
        if ( times.size()<n)
            times.push_back(duration);
        else{
            times[curr]=duration;
            curr++;
            if (curr>=times.size()) curr=0;}
    }

    double getAvrg()
    {
        double sum=0;
        for(auto t:times)
            sum+=t;
        return sum/double(times.size());
    }
}; …

我正在输入缺失值，missRanger因为我有 1000 个变量，所以花费的时间太长。我尝试使用并行计算，但它并没有使过程更快。这是代码

library(doParallel)
cores=detectCores()
cl <- makeCluster(cores[1]-1) 
registerDoParallel(cl)
library(missRanger)
train[1:lengthvar] <- missRanger(train[1:lengthvar], pmm.k = 3, num.trees = 100)
stopCluster(cl)

我不确定要向此代码添加什么才能使其正常工作。

这是我第一次尝试用 CUDA 实现递归。目标是使用 CUDA 的强大功能从一组字符“12345”中提取所有组合以动态并行化任务。这是我的内核：

__device__ char route[31] = { "_________________________"};
__device__ char init[6] = { "12345" };

__global__ void Recursive(int depth) {

    // up to depth 6
    if (depth == 5) return;     
    // newroute = route - idx

    int x = depth * 6;
    printf("%s\n", route);

    int o = 0;
    int newlen = 0;
    for (int i = 0; i<6; ++i)
    {
        if (i != threadIdx.x)
        {
            route[i+x-o] = init[i];
            newlen++;
        }
        else                
        {
            o = 1;
        }
    } …

我想连续运行n 个任务，但是这些任务是内存密集型的，我希望一次只有x 个处于活动状态。但最终所有这n 个任务都应该通过它们之间的上下文切换来运行。

简而言之，我想要另一种实现FixedThreadPool，其中额外的任务也应该与上下文切换一起运行。

我们是否有实现相同的线程池变体？或者有什么其他的实现方式？

更新：在阅读了一点并阅读下面的答案后，决定“分而治之”，即将连续运行的任务分解为小的短期任务单元，并一次又一次地提交给 FixedThreadPool