对 Amazon Elastic MapReduce 上的 R 映射器脚本进行故障排除 - 结果不符合预期

Question

我正在尝试使用 Amazon Elastic Map Red 运行一系列数百万个案例的模拟。这是一个没有减速器的 Rscript 流作业。我在 EMR 调用 --reducer org.apache.hadoop.mapred.lib.IdentityReducer 中使用 IdentityReducer。

当手动传递一行字符串时，脚本文件在测试时工作正常，并从 Linux 机器上的命令行本地运行 echo "1,2443,2442,1,5" | ./mapper.R 我得到了我所期望的一行结果。然而，当我使用 EMR 输入文件中的大约 10,000 个案例（行）测试我的模拟时，我只得到了 10k 输入行中十几行左右的输出。我已经尝试了好几次，但我无法弄清楚为什么。 Hadoop 作业运行良好，没有任何错误。似乎输入行被跳过，或者身份缩减器可能发生了一些事情。对于有输出的情况，结果是正确的。

我的输入文件是具有以下数据格式的 csv，由逗号分隔的一系列五个整数：

1,2443,2442,1,5
2,2743,4712,99,8
3,2443,861,282,3177
etc...

这是我的 ma​​pper.R 的 R 脚本

#! /usr/bin/env Rscript

# Define Functions
trimWhiteSpace <- function(line) gsub("(^ +)|( +$)", "", line)
splitIntoWords <- function(line) unlist(strsplit(line, "[[:space:]]+"))
# function to read in the relevant data from needed data files
get.data <- function(casename) {
    list <- lapply(casename, function(x) {
        read.csv(file = paste("./inputdata/",x, ".csv", sep = ""),
                 header = TRUE,
        stringsAsFactors = FALSE)})
    return(data.frame(list))
}

con <- file("stdin")            
line <- readLines(con, n = 1, warn = FALSE) 
line <- trimWhiteSpace(line)
values <- unlist(strsplit(line, ","))
lv <- length(values)
cases <- as.numeric(values[2:lv])
simid <- paste("sim", values[1], ":", sep = "")
l <- length(cases)                      # for indexing

## create a vector for the case names
names.vector <- paste("case", cases, sep = ".")

## read in metadata and necessary data columns using get.data function
metadata <- read.csv(file = "./inputdata/metadata.csv", header = TRUE,
                     stringsAsFactors = FALSE)
d <- cbind(metadata[,1:3], get.data(names.vector))

## Calculations that use df d and produce a string called 'output' 
## in the form of "id: value1 value2 value3 ..." to be used at a 
## later time for agregation.

cat(output, "\n")
close(con)

此模拟的（通用）EMR 调用是：

ruby elastic-mapreduce --create --stream --input s3n://bucket/project/input.txt --output s3n://bucket/project/output --mapper s3n://bucket/project/mapper.R --reducer org.apache.hadoop.mapred.lib.IdentityReducer --cache-archive s3n://bucket/project/inputdata.tar.gz#inputdata --name Simulation --num-instances 2

如果有人如果您对我可能遇到这些问题的原因有任何见解，我愿意接受建议以及对 R 脚本的任何更改/优化。

我的另一个选择是将脚本转换为函数并使用 R 多核包运行并行应用，但我还没有尝试过。我希望在 EMR 上实现此功能。我使用了 JD Long 的 和 原文

Answer 1

没有任何明显的东西跳出来。但是，您可以使用只有 10 行的简单输入文件来运行该作业吗？确保这 10 行是没有在您的大测试用例中运行的场景。尝试此操作以消除您的输入导致 R 脚本无法生成答案的可能性。

调试 EMR 作业本身就是一项技能。

编辑：

这是一次彻底的钓鱼探险，但使用 AWS GUI 启动 EMR 交互式猪会话。 “交互式猪”会话保持正常运行，以便您可以通过 ssh 访问它们。您也可以从命令行工具执行此操作，但从 GUI 中执行此操作会更容易一些，因为您只需执行一次即可。然后 ssh 进入集群，传输测试用例 infile 缓存文件和映射器，然后运行以下

命令：cat infile.txt | yourMapper.R> outfile.txt

这只是为了测试您的映射器是否可以在 EMR 环境中解析 infile，而无需使用 Hadoop 位。

编辑2：

我将上面的文本留在那里供后代使用，但真正的问题是你的脚本永远不会返回到标准输入来获取更多数据。因此，每个映射器都会运行一次，然后结束。如果运行上述一行，您将只能得到一个结果，而不是 infile.txt 中每一行的结果。如果您甚至在本地计算机上运行了 cat 测试，则应该会弹出错误！

让我们看看 Pete 的 R 示例中的字数：

#! /usr/bin/env Rscript

trimWhiteSpace <- function(line) gsub("(^ +)|( +$)", "", line)
splitIntoWords <- function(line) unlist(strsplit(line, "[[:space:]]+"))

## **** could wo with a single readLines or in blocks
con <- file("stdin", open = "r")
while (length(line <- readLines(con, n = 1, warn = FALSE)) > 0) {
    line <- trimWhiteSpace(line)
    words <- splitIntoWords(line)
    ## **** can be done as cat(paste(words, "\t1\n", sep=""), sep="")
    for (w in words)
        cat(w, "\t1\n", sep="")
}
close(con)

您的脚本缺少的部分是这样的：

 while (length(line <- readLines(con, n = 1, warn = FALSE)) > 0) {
        #do your dance
        #do your dance quick
        #come on everybody tell me what's the word
        #word up
    }

你自然应该替换 Cameo 的 Word Up 的歌词！与你的实际逻辑。

请记住，正确的调试音乐可以减轻整个过程的痛苦：

http://www.youtube.com/观看？v=MZjAantupsA