标签: distributed-computing

每当我在S-Plus中进行大规模的蒙特卡罗模拟时,我总是在等待它完成时留下胡须.

在R中运行蒙特卡罗模拟的最佳技巧是什么？以分布式方式运行流程的任何好例子？

我想知道如果OpenMPI/MPICH2集群的节点终止会发生什么？是否有一些机制可以容忍这种情况并继续执行？

谢谢你的回答海因里希

假设有人建立了一个cassandra集群.你有一个10 [TB]数据库,它在10个节点之间均匀分布,一切运行顺畅等.

假设您有100台计算机可供使用,每台计算机都尝试从cassandra集群中读取(不同的)数据.此外,您有许多不断需要运行的作业,每个作业都在不同的时间(显然,每个作业都需要在不同的机器上运行).

你如何管理所有这些任务/工作？你如何在机器之间分配任务？你如何跟踪过程中的工作/机器？

是否有任何开源工具(最好是有Python客户端)可以帮助在Linux环境中进行操作？

我目前正在Java中使用SSH库,但它似乎缺乏执行多部分命令的能力(例如,如果我passwd user没有办法输入密码两次以将其更改为,因为它会让你开始新的每次输入命令时的会话).我真的需要这个功能用于我正在使用的当前应用程序,并且似乎有很多用于Java的SSH库,但我不确定哪些会允许这样,因为有些人似乎没有.

任何关于一个好的图书馆的建议都会非常感激.

对于近距离巡逻:如果你要尝试关闭它,至少引用一个线程,询问与我相同的问题,而不仅仅是一般的"什么是一个好的Java ssh库"问题

我在最近的一次采访中遇到了这个问题:

你有一个范围内的输入数字流,0 to 60000你有一个函数,它将从该范围获取一个数字,并返回该数字的出现次数,直到那一刻.提供合适的数据结构/算法来实现该系统.

我的解决方案是:

创建一个大小为60001的数组,指向位向量.这些位向量将包含传入数字的计数,并且传入的数字也将用于索引相应数字的数组.随着计数变得太大而不能容纳它们,位向量将动态增加.

因此,如果数字100numbers/sec按此速率进行,则在1百万年内,总数将为= (100*3600*24)*365*1000000 = 3.2*10^15.在最坏的情况下,流中的所有数字都是相同的ceil((log(3.2*10^15) / log 2) )= 52bits,如果数字是均匀分布的(3.2*10^15) / 60001 = 5.33*10^10,那么每个数字的出现次数将需要36 bits每个数字的总数.因此,假设4byte指针我们需要(60001 * 4)/1024 = 234 KB数组的内存,对于具有相同数字的情况,我们需要位向量大小= 52/8 = 7.5 bytes仍然大约234KB.而对于另一种情况,我们需要(60001 * 36 / 8)/1024 = 263.7 KB位向量总计约500KB.因此,用普通的PC和内存来做这件事是非常可行的.

但是访问者说,因为它是无限的流,它最终会溢出并给我提示,如果有很多PC我们怎么能这样做,我们可以在它们之间传递消息或考虑文件系统等.但我一直在想是否这个解决方案当时没有工作,其他人也会这样.不用说,我没有得到这份工作.

如何用更少的内存来解决这个问题？你能想到另一种方法(使用PC网络)吗？

我有兴趣在RabbitMQ中实现"工作队列"模型.但是,我发现代理执行简单的循环操作,将任务分配给工作人员.

https://www.rabbitmq.com/tutorials/tutorial-two-java.html

如果某个特定的工作人员忙于执行非常繁重的任务并且还有其他自由工作者,则代理应该能够将队列中的消息分发给下一个可用的工作人员,而不是循环序列中的下一个工作人员.有没有办法使用RabbitMQ实现这一目标？

我试图在一个小型独立集群(1个主节点和8个从节点)上设置Apache-Spark.我已经安装了基于Hadoop 2.4构建的spark 1.1.0的"预构建"版本.我在节点之间设置了无密码ssh并导出了一些必要的环境变量.其中一个变量(可能最相关)是:

export SPARK_LOCAL_DIRS=/scratch/spark/

我有一小段python代码,我知道它与Spark一起使用.我可以在本地运行 - 在我的桌面上,而不是群集上 - 使用:

$SPARK_HOME/bin/spark-submit ~/My_code.py

我将代码复制到了集群.然后,我从头节点开始所有进程:

$SPARK_HOME/sbin/start-all

并且每个从属列为作为进程xxxxx运行.

如果我然后尝试使用上面相同的命令运行我的代码:

$SPARK_HOME/bin/spark-submit ~/MY_code.py

我收到以下错误:

14/10/27 14:19:02 ERROR util.Utils: Failed to create local root dir in /scratch/spark/.  Ignoring this directory.
14/10/27 14:19:02 ERROR storage.DiskBlockManager: Failed to create any local dir.

我在/scratch和/scratch/spark777 上设置了权限.非常感谢任何帮助.

我尝试使用Supercomputers的安装脚本在Cluster系统上安装pyCOMPSs(v1.4).该脚本终止时出现以下错误:

libtool: link: ranlib .libs/libcbindings.a
libtool: link: ( cd ".libs" && rm -f "libcbindings.la" && ln -s
"../libcbindings.la" "libcbindings.la" )
make[1]: Entering directory
`/home/xxx/repos/pycompss/COMPSs/Bindings/c/src/bindinglib'
  /usr/bin/mkdir -p
'/home/cramonco/svn/compss/framework/trunk/builders/specs/deb/compss-c-binding/tmp/opt/COMPSs/Bindings/c/lib'
/usr/bin/mkdir: cannot create directory ‘/home/cramonco’: Permission denied
make[1]: *** [install-libLTLIBRARIES] Error 1
make[1]: Leaving directory
`/home/xxx/xxx/repos/pycompss/COMPSs/Bindings/c/src/bindinglib'
make: *** [install-am] Error 2

BindingLib Installation failed, please check errors above!

我已经dask在我的集群上运行,但我似乎无法访问诊断网页.着陆页可见,如下所示:

但是所有链接都挂起并且从不加载页面.

调度程序在此输出时正常启动:

[hoffmand@h05u06 ~]$ dask-scheduler --scheduler-file dask-scheduler.json
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO -   Scheduler at:   tcp://10.36.105.16:8786
distributed.scheduler - INFO -       bokeh at:              0.0.0.0:8788
distributed.scheduler - INFO -        http at:              0.0.0.0:9786
distributed.bokeh.application - INFO - Web UI: http://127.0.0.1:8787/status/
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO - Register tcp://10.36.107.15:37780
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.36.107.15:37780

我了解tensorflow分布式培训，并且实现了自己的脚本。

我现在想要做的是整合为某些工作人员分配异步评估模型任务的可能性。

假设我们有6个工作人员，我要做的是使用其中4个工作人员进行异步培训，一个工作人员定期评估模型，另一个工作人员定期进行模型推断。

我的直觉是实现以下目标：

....
elif FLAGS.job_name == "worker":

    if FLAGS.task_index <= (len(cluster_dict["worker"][:-2]) - 1):
         logging.info("Training worker started")
         ...
        with tf.device(tf.train.replica_device_setter(
                worker_device="/job:worker/task:%d" % FLAGS.task_index,
                cluster=cluster,
                ps_tasks=len(cluster_dict["ps"])
            )):
                train_model = Model(
                    mode=tf.contrib.learn.ModeKeys.TRAIN
                )
               with tf.train.MonitoredTrainingSession(
                    is_chief=(FLAGS.task_index == 0),
                    master=server.target,
                    checkpoint_dir=ckpt_dir,
                    config=config_proto,
                    hooks=hooks
                ) as mon_sess:
                    while not mon_sess.should_stop():
                        res = train_model.train(...)
                        ...

   elif FLAGS.task_index == (len(cluster_dict["worker"][-2]) - 1):
         logging.info("Evaluation worker started")
         ...
         with tf.device(tf.train.replica_device_setter(
              worker_device="/job:worker/task:%d" % FLAGS.task_index,
              cluster=cluster,
              ps_tasks=len(cluster_dict["ps"])
          )):
              eval_model = Model(
                 mode=tf.contrib.learn.ModeKeys.EVAL
              )
              ...

   elif FLAGS.task_index == (len(cluster_dict["worker"][-1]) - …