使用RevoScaleR的每组行数

Question

我正在转换一个本地R脚本，以利用革命-R(也就是Microsoft客户机/服务器)包中的RevoScaleR函数。这是为了能够更好地扩大与大量的数据。

目标是创建一个新列，将每个组的行编号。使用data.table可以使用以下代码实现这一点：

library(data.table)
eventlog[,ActivityNumber := seq(from=1, to=.N, by=1), by=Case.ID]

为了便于说明，输出如下：

    Case.ID    ActivityNumber
1       A              1
2       A              2
3       B              1
4       C              1
5       C              2
6       C              3

在使用rx-functions进行了一些研究之后，我发现了包dplyrXdf，它基本上是在Xdf存储的数据上使用dplyr函数的包装器，同时仍然受益于RevoScaleR的优化函数(参见http://blog.revolutionanalytics.com/2015/10/using-the-dplyrxdf-package.html)。

就我的情况而言，这将导致以下情况：

result <- eventlog %>%
  group_by(Case.ID) %>%
  mutate(ActivityNumber = seq_len(n()))

但是，这会导致以下错误：

ERROR: Attempting to add a variable without a name to an analysis.
Caught exception in file: CxAnalysis.cpp, line: 3756. ThreadID: 1248 Rethrowing.
Caught exception in file: CxAnalysis.cpp, line: 5249. ThreadID: 1248 Rethrowing.
Error in doTryCatch(return(expr), name, parentenv, handler) : 
  Error in executing R code: ERROR: Attempting to add a variable without a name to an analysis.

有什么办法解决这个错误吗？或者其他(更好的？)获取所请求的结果的方法？

Answer 1

感谢@马特-帕克为我指出了这个问题。

注意，n()不是一个常规的R函数，尽管它看起来像一个。它需要为每个数据源特别实现，可能还需要为每个mutate、summarise和filter分别实现。

现在，xdf文件支持的n的唯一用法是在summarise中计算行数。为其他动词实现它实际上是非常重要的。

特别是，Matt使用seq_along实现n的功能存在一个问题。请记住，xdf文件是块结构的:每一块行都是独立于其他块读取和处理的。这意味着生成的序列仅针对该行块，而不是针对一个组中的所有行。如果一个组跨越多个块，序列号将在中间重新启动。

获取正确序列号的方法是，对该组读取的行数进行运行计数，并在每次处理块时对其进行更新。您可以使用transformFunc来完成这一任务，通过.rxArgs参数传递给transmute：

ev <- eventlog %>% group_by(Case.ID) %>% transmute(.rxArgs = list(
    transformFunc = function(varList) {
        n <- .n + seq_along(varList[[1]])
        if(!.rxIsTestChunk)  # need this b/c rxDataStep does a test run on the 1st 10 rows
            .n <<- n[length(n)]
        list(n=n)
    },
    transformObjects = list(.n = 0))

这应该适用于local、localpar和foreach计算上下文。在无法保证rxDataStep按确定的顺序处理行的任何上下文中(或者至少不会给出一个可重复的结果)，它可能不起作用--所以Mapreduce、Spark、Teradata或类似的。

Answer 2

我不知道为什么这样做，但尝试使用seq_along(Case.ID)而不是seq_len(n())

result <- eventlog %>%
  group_by(Case.ID) %>%
  mutate(ActivityNumber = seq_along(Case.ID))

这似乎是n()的一些问题。下面是我的探索性代码，以防其他人想实验：

options(stringsAsFactors = FALSE)

library(dplyrXdf)

# Set up some test data
eventlog_df <- data.frame(Case.ID = c("A", "A", "A", "A", "A", "B", "C", "C", "C"))

# Add a variable for artificially splitting the XDF into small chunks
eventlog_df$Chunk.ID <- factor((seq_len(nrow(eventlog_df)) + 2) %/% 3)

# Check the results
eventlog_df


# Now read it into an XDF file. I'm going to read just three rows in at a time
# so that the XDF file has several chunks, so we can be confident this works
# across chunks

eventlog <- tempfile(fileext = ".xdf")

for(i in 1:3) {
    rxImport(inData = eventlog_df[eventlog_df$Chunk.ID %in% i, ],
             outFile = eventlog,
             colInfo = list(Case.ID = list(type = "factor", 
                                           levels = c("A", "B", "C"))),
             append = file.exists(eventlog))
}

# Convert to a proper data source
eventlog <- RxXdfData(eventlog)

rxGetInfo(eventlog, getVarInfo = TRUE, numRows = 10)


# Now to dplyr. First, let's make sure it can count up the records
# in each group without any trouble.
result <- eventlog %>%
  group_by(Case.ID) %>%
  summarise(ActivityNumber = n())

# It can:
rxDataStep(result)


# Now if we switch to mutate, does n() still work?
result <- eventlog %>%
  group_by(Case.ID) %>%
  mutate(ActivityNumber = n())

# No - and it seems to be complaining about missing variables. So what if
# we try to refer to a variable we *know* exists?
result <- eventlog %>%
  group_by(Case.ID) %>%
  mutate(ActivityNumber = seq_along(Case.ID))

# It works
rxDataStep(result)

Answer 3

dplyr和dplyrXdf有一个tally方法，用于对每个组的项进行计数：

result <- eventlog %>%
  group_by(Case.ID) %>%
  tally()

如果您想做的不仅仅是列出每组的记录，您可以使用汇总(因为您没有显示数据，所以我使用了一个名为delay的假设列，我假设它是数字，用于说明性目的)：

result <- eventlog %>%
  group_by(Case.ID) %>%
  summarize(counts = n(),
            ave_delay = mean(delay))

您可以使用常规的RevoScaleR函数来完成上述操作，

rxCrossTabs(~ Case.ID, data = eventlog)

第二个例子是：

rxCube(delay ~ Case.ID, data = eventlog)