标签: parallel-processing

场景:接收数据并将其写入带有时间戳的数据库.我需要按照基于时间戳接收的顺序处理原始数据,并将其写回数据库,不同的表,再次根据时间戳维护订单.

我提出了以下设计:创建了两个队列,一个用于存储来自数据库的原始数据,另一个用于存储已处理的数据,然后再写回数据库.我有两个线程,一个读取初始队列,另一个读取Result队列.在我之间生成多个线程来处理来自Initial队列的数据并将其写入Result队列.

我已经尝试过SortedList(手动锁定)和BlockingCollection.我使用了两种并行处理方法:Parallel.For(ForEach)和TaskFactory.Task.StartNew.

基于若干因素,每个数据单元可能花费可变的时间来处理.一个线程仍然可以处理第一个数据点,而其他线程每个处理三个或四个数据点,搞乱了时间戳顺序.

我最近发现了OrderingPartitioner,我认为它可以解决问题,但是按照MSDN的例子我可以看到,它也没有对基础集合进行排序.可能是我需要实现自定义分区器来订购我的复杂数据类型集合？或者可能有更好的方法来解决问题？

任何有关讨论类似问题的文章的建议和/或链接都受到高度赞赏.

我有一个四核CPU.我创建了4个线程并运行了一个cpu密集型循环,它比在一个线程中以程序方式运行所需的时间长4倍.

我创建了两个要比较的项目,一个是线程,另一个没有.我将展示代码和运行时间.请注意没有线程的项目看起来很奇怪的原因是我想复制内存开销,因为我不确定它会影响运行时间.所以,这是没有线程的代码:

class TimeTest implements Runnable {
    private Thread t;
    private String name;

    TimeTest(String name) {
        this.name = name;
        System.out.println("Creating class " + name);
    }

    public void run() {
        System.out.println("Running class " + name);

        int value = 100000000;
//        try {
            while (--value > 0) {
                Math.random();
//                Thread.sleep(1);
//                System.out.println("Class " + name + " " + value);
            }
//        } catch (InterruptedException e) {
//            System.out.println("Interrupted " + name);
//        }

        System.out.println("Class " + name + " exiting..."); …

在这里,我使用Javaparallel流来迭代List并使用每个列表元素作为输入调用REST调用.我需要将REST调用的所有结果添加到我正在使用的集合中ArrayList.下面给出的代码工作正常,只是ArrayList的非线程安全性会导致错误的结果,并且添加所需的同步会导致争用,从而破坏并行性的好处.

有人可以建议我在我的案例中使用并行流的正确方法.

public void myMethod() {
    List<List<String>> partitions = getInputData();
    final List<String> allResult = new ArrayList<String>();
    partitions.parallelStream().forEach(serverList -> callRestAPI(serverList, allResult);
}

private void callRestAPI(List<String> serverList, List<String> allResult) {
    List<String> result = //Do a REST call.
    allResult.addAll(result);
}

使用Spring批处理,我希望我的步骤可以跨节点分布,并为给定的作业执行它们.我有一个用例,其中一个作业有多个步骤,每个步骤可以在托管应用程序的多个节点中运行.有人试过吗？任何想法都将受到高度赞赏!

我正在尝试同时学习Perl 6和并行/并发.

对于一个简单的学习练习,我有一个550'.htm'文件的文件夹,我想要所有这些文件中的代码行总和.到目前为止,我有这个:

use v6;

my $start_time = now;
my $exception;
my $total_lines = 0;

my @files = "c:/testdir".IO.dir(test => / '.' htm $/);
for @files -> $file {
    $total_lines += $file.lines.elems;
    CATCH {
        default { $exception = $_; } #some of the files error out for malformed utf-8
    }
}
say $total_lines;
say now - $start_time;

这在大约3秒内得到577,449的总和.

我如何重写它以利用Perl 6并行思想？我意识到节省的时间不会太多,但它可以作为概念的证明.

我是新手,并且误解了Gunicorn + Flask的工作方式.

当我使用4名工作人员运行Gunicorn时,它会创建我的Flask应用程序的4个实例,或者它将创建4个处理来自Nginx和Flask应用程序实例的Web请求的进程？

如果我在我的应用程序中简单地实现内存缓存(例如字典),那么gunicorn会创建多个app实例,因此会创建多个缓存实例吗？

我想使用批处理将工作从主服务器分发到多个工作服务器.

理想情况下,我会有一个tasks.txt文件,其中包含要执行的任务列表

cmd args 1
cmd args 2
cmd args 3
cmd args 4
cmd args 5
cmd args 6
cmd args 7
...
cmd args n

并且每个工作服务器将使用ssh进行连接,读取文件并将每一行标记为正在进行或已完成

#cmd args 1  #worker1 - done
#cmd args 2  #worker2 - in progress
#cmd args 3  #worker3 - in progress
#cmd args 4  #worker1 - in progress 
cmd args 5
cmd args 6
cmd args 7
...
cmd args n

我知道如何进行ssh连接,读取文件,远程执行,但不知道如何进行读写原子操作,以免出现2台服务器启动相同任务的情况,以及如何更新这条线.

我希望每个工作人员都能进入任务列表并锁定列表中的下一个可用任务,而不是服务器主动命令工作人员,因为我将根据如何启动或关闭灵活数量的工作克隆我需要完成任务.

更新:

我对工人脚本的想法是:

#!/bin/bash

taskCmd=""
taskLine=0
masterSSH="ssh usr@masterhost"
tasksFile="/path/to/tasks.txt" …

我在OpenMPHyperThreaded CPU上使用代码.

如果其他条件相同,那么非HyperThreaded CPU的性能会如何变化？

我注意到100%的处理器利用率,无论我运行多少线程,但改变线程数确实提高了性能.怎么会这样？

非INTEL多线程CPU的故事是否相同？

我正在研究并行编程概念，并尝试在单核上优化矩阵乘法示例。到目前为止，我想到的最快的实现是：

/* This routine performs a dgemm operation
 *  C := C + A * B
 * where A, B, and C are lda-by-lda matrices stored in column-major format.
 * On exit, A and B maintain their input values. */    
void square_dgemm (int n, double* A, double* B, double* C)
{
  /* For each row i of A */
  for (int i = 0; i < n; ++i)
    /* For each column j of B */
    for (int j = …

我是一个火花新手,并使用Spark SQL/hiveContext有一个简单的spark应用程序:

1. 从蜂巢表中选择数据(10亿行)
2. 做一些过滤,聚合包括row_number over window function来选择第一行,group by,count()和max()等.
3. 将结果写入HBase(数亿行)

我提交作业在纱线集群(100个执行器)上运行它,它很慢,当我在Spark UI中查看DAG可视化时,似乎只有hive表扫描任务并行运行,其余步骤#2和#以上3只在一个实例中运行,可能应该能够优化并行化？

该应用程序看起来像:

步骤1:

val input = hiveContext
  .sql(
     SELECT   
           user_id  
           , address  
           , age  
           , phone_number  
           , first_name  
           , last_name  
           , server_ts   
       FROM  
       (     
           SELECT  
               user_id  
               , address  
               , age  
               , phone_number  
               , first_name  
               , last_name  
               , server_ts   
               , row_number() over 
                (partition by user_id, address,  phone_number, first_name, last_name  order by user_id, address, phone_number, first_name, last_name,  server_ts desc, age) AS rn  
           FROM  
           (  
               SELECT  
                   user_id  
                   , address  
                   , age  
                   , phone_number …