R 连接到 EC2 实例以进行并行处理

Question

我在初始化从 R 到 AWS EC2 实例的连接时遇到问题，因为我似乎不断收到错误消息：Permission denied (publickey) 我目前正在使用 Mac OS X 10.6.8 作为我的操作系统

我尝试在终端（$）然后R（>）中运行的代码如下：

$ R --vanilla
> require(snowfall)
> sfInit(parallel=TRUE,socketHosts =list("ec2-xx-xxx-xx-xx.zone.compute.amazonaws.com"))
Permission denied (publickey)

但奇怪的是，当我尝试 ssh 进入实例时我不需要密码，因为我已经在初始化时将公钥导入到实例中，（我认为）

所以从我的普通终端...运行时

$ ssh ubuntu@ec2-xx-xxx-xx-xx.zone.compute.amazonaws.com

它会自动连接...（所以我不能 100% 确定它是否像 Using snow (and snowfall) with AWS for parallel processing in R 那样是无密码问题）

我已经尝试过查看大量关于钥匙等的材料，但似乎没有一个有太大的不同。此外，由于某种原因，我的~/.ssh/authorized_keys 是一个文件夹而不是文件，即使尝试sudo cd .ssh/authorized_keys...我也无法访问它...就权限而言，它具有drw-------

最终目标是连接到很多 ec2 实例并使用foreach 进行一些并行处理...但是现在连接到一个会很好...我也想使用我自己的 ami所以星团并不是我真正想要的......（除非我能够使用私有 amis 并私下运行所有​​命令......）

如果 doRedis 比有人能告诉我如何从本地机器连接到 ec2 实例更好的话，那也很好......

编辑

我已经设法使用parallel 包makePSOCKcluster 处理ssh 无密码登录，如R and makePSOCKcluter EC2 socketConnection 所示......但现在遇到socketConnection 问题，如链接中的问题所示。 ..

任何想法如何连接到它？

也证明一切正常，我猜这意味着以下命令/函数可以获取所有不同的 IP 地址

d <- parLapply(cl1, 1:length(cl1),function(x)system("ifconfig",intern=T)[2])

其中cl1 是make*cluster 函数的输出

注意，因为赏金实际上是针对链接中的问题....我不介意您发布答案的问题...但是只要写了一些东西在这个问题上将其链接到链接问题的正确答案，然后我将相应地奖励积分......

Answer 1

在尝试将主节点保持在本地时，我在并行 EC2 设置方面也遇到了很多问题。使用StarCluster 设置池有很大帮助，但真正的改进来自于使用 StarCluster和在 EC2 私有 IP 池中拥有主节点。

StarCluster 为所有节点以及使用的任何挂载设置所有密钥处理。动态节点分配不可行，但除非要长期使用 Spot 实例并且您的出价策略不能“保留”您的实例，否则动态分配应该是一个问题。

其他一些经验教训：

• 创建一个包含私有 IP 的变量以传递给 createCluster 并将其导出，因此当您需要使用相同的节点重新启动时，它会更容易。
• 让主节点运行 byobu 并将其设置为 R 会话日志记录。
• 在主节​​点上运行 RStudio 服务器有时会很有帮助，但应该是与从节点不同的 AMI。 :)
• 让控制脚本将数据 rda 文件卸载到远程监控新文件的路径并自动下载它们。
• 使用 htop 监控从属服务器，以便您轻松查看实例并确定脚本要求（内存/cpu/可扩展性）。
• 利用处理器超线程启用/禁用脚本。

我在从属连接和序列化/反序列化方面遇到了很多问题，发现其中一件事是连接限制，连接限制需要通过节点数来减少；当控制脚本停止时，最简单的清理方法是重新启动主 R 会话，并使用脚本杀死从属进程而不是等待超时。

设置确实需要一些工作，但希望这些想法对您有所帮助...

---

虽然是 8 个月前，StarCluster 和 R 都发生了变化，但这里有一些设置方式……您可以在 StarCluster 文档中找到 90% 的内容。

• 根据来自 AWS 控制台的安全信息设置 .starcluster/config AWS 和密钥对部分。
• 定义 [smallcluster]
  • 键名
  • 可用区
• 定义扩展 [smallcluster] 的集群模板。使用基于 StarCluster 64 位 HVM AMI 的 AMI。我没有创建新的公共 AMI 实例，而是保存了一个已配置的实例（包含我需要的所有工具）并将其用作 AMI。

这是一个例子......

[cluster Rnodes2]
EXTENDS=smallcluster
MASTER_INSTANCE_TYPE = cc1.4xlarge
MASTER_IMAGE_ID= ami-7621f91f
NODE_INSTANCE_TYPE = cc2.8xlarge
NODE_IMAGE_ID= ami-7621f91f
CLUSTER_SIZE= 8
VOLUMES= rdata
PLUGINS= pkginstaller
SPOT_BID= 1.00

• 设置共享卷，这是屏幕/byoubu 日志、主要 .R 脚本检查点输出、共享 R 数据以及生产包的源。它会在名为 export 的子路径中监视新文件，因此如果集群或控制脚本死亡/异常，最大数量的记录将全部丢失并需要重新计算。

创建共享卷后，定义很简单：

[volume rdata]
VOLUME_ID = vol-1145497c
MOUNT_PATH = /rdata

确保所有节点上的最新（和相同）R 版本的包安装程序。

[plugin pkginstaller]
setup_class = starcluster.plugins.pkginstaller.PackageInstaller
packages = r-base, r-base-dev, r-recommended

最后，ssh 和 RStudio 服务器的访问权限。通过代理的 Https 会更安全，但由于 RStudio 仅用于控制脚本设置...

[permission ssh]
# protocol can be: tcp, udp, or icmp
protocol = tcp
from_port = 22
to_port = 22

# [permission http]
protocol = tcp
from_port = 8787
to_port = 8787

然后使用 StarCluster 界面启动集群。它处理所有的访问控制、系统名称、共享等...一旦集群运行，我从本地系统运行一个 ssh 会话，并运行一个脚本来停止超线程：

#!/bin/sh

# disable hyperthreading
for cpunum in $(
    cat /sys/devices/system/cpu/cpu*/topology/thread_siblings_list | 
    cut -s -d, -f2- | tr ',' '\n' | sort -un); do
        echo 0 > /sys/devices/system/cpu/cpu$cpunum/online
done

然后在每个节点上启动一个 htop 会话，以根据导出的检查点日志监控可扩展性。

然后，登录到主服务器，开始一个屏幕会话（从那以后我更喜欢 byobu）并从 StarCluster 安装的卷中启动 R。这样，当集群由于某种原因停止时，我可以通过启动 R 轻松地再次设置。一旦进入 R，第一件事就是使用 nodeXXX 名称创建一个 workers.list 变量，这很简单：

cluster.nodes <- c("localhost", paste("node00", 1:7, sep='' ) )
workers.list <- rep( cluster.nodes, 8 )

然后我加载了控制脚本，退出并保存了工作区。控制脚本处理所有用于导出和检查点的表输出以及对生产包的包装调用。脚本的 main 函数还接受了一个 cpus 参数，这是放置工作人员列表的位置，然后将其作为 cores 传递给集群初始化程序。

initialize.cluster <- function( cores )
{
  if( exists( 'cl' ) ) stopCluster( cl )

  print("Creating Cluster")
  cl <- makePSOCKcluster( cores )    
  print("Cluster created.")
  assign( 'cl', cl, envir=.GlobalEnv )
  print( cl )

  # All workers need to have the bounds generator functions...
  clusterEvalQ( cl, require('scoreTarget') )
  # All workers need to have the production script and package.
  clusterExport( cl, varlist=list('RScoreTarget', 'scoreTarget'))
  return ( cl )
}

一旦 R 会话重新启动（在最初创建 worker.list 之后），就会获取控制脚本，并调用 main 函数。就是这样。有了这个设置，如果集群停止了，我就退出主主机上的 rsession；在每个从站上通过 htop 停止从站进程并重新启动。

这是一个实际的例子::

R

R version 2.15.0 (2012-03-30)
Copyright (C) 2012 The R Foundation for Statistical Computing
ISBN 3-900051-07-0
Platform: x86_64-pc-linux-gnu (64-bit)

R is free software and comes with ABSOLUTELY NO WARRANTY.
You are welcome to redistribute it under certain conditions.
Type 'license()' or 'licence()' for distribution details.

  Natural language support but running in an English locale

R is a collaborative project with many contributors.
Type 'contributors()' for more information and
'citation()' on how to cite R or R packages in publications.

Type 'demo()' for some demos, 'help()' for on-line help, or
'help.start()' for an HTML browser interface to help.
Type 'q()' to quit R.

[Previously saved workspace restored]

> source('/rdata/buildSatisfactionRangeTable.R')
Loading required package: data.table
data.table 1.7.7  For help type: help("data.table")
Loading required package: parallel
Loading required package: scoreTarget
Loading required package: Rcpp
> ls()
 [1] "build.satisfaction.range.table" "initialize.cluster"            
 [3] "initialize.table"               "parallel.choices.threshold"    
 [5] "rolled.lower"                   "rolled.upper"                  
 [7] "RScoreTarget"                   "satisfaction.range.table"      
 [9] "satisfaction.search.targets"    "search.range.bound.offsets"    
[11] "search.range.bounds"            "search.range.center"           
[13] "Search.Satisfaction.Range"      "update.bound.offset"           
[15] "workers.list"                  
> workers.list
  [1] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
  [7] "localhost" "localhost" "node001"   "node002"   "node003"   "node004"  
 [13] "node005"   "node006"   "node007"   "node001"   "node002"   "node003"  
 [19] "node004"   "node005"   "node006"   "node007"   "node001"   "node002"  
 [25] "node003"   "node004"   "node005"   "node006"   "node007"   "node001"  
 [31] "node002"   "node003"   "node004"   "node005"   "node006"   "node007"  
 [37] "node001"   "node002"   "node003"   "node004"   "node005"   "node006"  
 [43] "node007"   "node001"   "node002"   "node003"   "node004"   "node005"  
 [49] "node006"   "node007"   "node001"   "node002"   "node003"   "node004"  
 [55] "node005"   "node006"   "node007"   "node001"   "node002"   "node003"  
 [61] "node004"   "node005"   "node006"   "node007"   "node001"   "node002"  
 [67] "node003"   "node004"   "node005"   "node006"   "node007"   "node001"  
 [73] "node002"   "node003"   "node004"   "node005"   "node006"   "node007"  
 [79] "node001"   "node002"   "node003"   "node004"   "node005"   "node006"  
 [85] "node007"   "node001"   "node002"   "node003"   "node004"   "node005"  
 [91] "node006"   "node007"   "node001"   "node002"   "node003"   "node004"  
 [97] "node005"   "node006"   "node007"   "node001"   "node002"   "node003"  
[103] "node004"   "node005"   "node006"   "node007"   "node001"   "node002"  
[109] "node003"   "node004"   "node005"   "node006"   "node007"   "node001"  
[115] "node002"   "node003"   "node004"   "node005"   "node006"   "node007"  
> build.satisfaction.range.table(500000, FALSE, workers.list )
[1] "Creating Cluster"
[1] "Cluster created."
socket cluster with 120 nodes on hosts ‘localhost’, ‘node001’, ‘node002’, ‘node003’, ‘node004’, ‘node005’, ‘node006’, ‘node007’
Parallel threshold set to: 11000 
Starting at: 2 running to: 5e+05 :: Sat Apr 14 22:21:05 2012 

---

如果您已阅读此处，那么您可能有兴趣知道我测试了我可以测试的每个集群设置（包括 openMPI），发现速度没有差异，这可能是因为我的计算受 CPU 限制，也许不是。

此外，即使开始使用 HPC 会很痛苦，也不要放弃。这完全值得。如果我在商品工作站上坚持使用 base-R 中的幼稚实现，我仍然会等待完成我正在运行的计算的前 100,000 次迭代（好吧，不是真的，因为我永远不会坚持使用 R :D ）。借助集群，一周内完成了 384,000 次迭代。完全值得花时间（而且花了很多时间）来设置。