2

在我的 R 脚本中，我使用 pmdplyr 函数mutate_cascade()来tlag()改变我的数据，其中包含超过 300 万条记录，因此代码非常慢，但它可以工作。为了加快速度，我尝试添加 multidplyr 的并行处理功能。但它抛出了错误：小标题中的所有列都必须是向量。x 列.是一个multidplyr_party_df对象。 那是因为不可能在 multidplyr 集群上运行 pmdplyr pibble 吗？我是 pmdplyr 和 multidplyr 的新手，所以也许我只是做错了什么？

我得到一个import_data带有变量的合并数据集uuid, location_id, import_date, customer_name, total_value。异常导入可能会导致巨大的峰值，total_value因此我的代码试图平衡这个值的极不可能的波动（这是相对于每个客户的）：


cluster_library(cluster, "dplyr")
cluster_library(cluster, "pmdplyr")

mydata <- import_data %>%

  mutate(
    time_var = time_variable(import_date),
    id_var = id_variable(uuid, location_id),
    total_value_imported = total_value
  ) %>%

  arrange(uuid, location_id, import_date) %>%

  group_by(uuid, location_id) %>%

  partition(cluster) %>%

  pibble(
    .i = id_var,
    .t = time_var,
    .d = 0
  ) %>%

  mutate_cascade(
    total_value = case_when(
      total_value > (tlag(total_value, .n = 1, .quick = TRUE) * 10)  
        ~ tlag(total_value, .n = 1, .quick = TRUE),
      total_value < (tlag(total_value, .n = 1, .quick = TRUE) / 10) 
        ~ tlag(total_value, .n = 1, .quick = TRUE),
      TRUE ~ total_value_imported
    )
  ) %>%

  collect()

> rlang::last_error()
<error/tibble_error_column_scalar_type>
All columns in a tibble must be vectors.
x Column `.` is a `multidplyr_party_df` object.
Backtrace:
  1. dplyr::mutate(...)
  1. dplyr::arrange(., uuid, location_id, import_date)
  1. dplyr::group_by(., uuid, location_id)
  1. multidplyr::partition(., cluster)
  8. pmdplyr::pibble(., .i = id_var, .t = time_var, .d = 0)
  9. tibble::tibble(...)
 10. tibble:::tibble_quos(xs[!is_null], .rows, .name_repair)
 11. tibble:::check_valid_col(res, col_names[[j]], j)
 12. tibble:::check_valid_cols(set_names(list(x), name))
Run `rlang::last_trace()` to see the full context.
> rlang::last_trace()
<error/tibble_error_column_scalar_type>
All columns in a tibble must be vectors.
x Column `.` is a `multidplyr_party_df` object.
Backtrace:
     █
  1. └─`%>%`(...)
  2.   ├─base::withVisible(eval(quote(`_fseq`(`_lhs`)), env, env))
  3.   └─base::eval(quote(`_fseq`(`_lhs`)), env, env)
  4.     └─base::eval(quote(`_fseq`(`_lhs`)), env, env)
  5.       └─`_fseq`(`_lhs`)
  6.         └─magrittr::freduce(value, `_function_list`)
  7.           └─function_list[[i]](value)
  8.             └─pmdplyr::pibble(., .i = id_var, .t = time_var, .d = 0)
  9.               └─tibble::tibble(...)
 10.                 └─tibble:::tibble_quos(xs[!is_null], .rows, .name_repair)
 11.                   └─tibble:::check_valid_col(res, col_names[[j]], j)
 12.                     └─tibble:::check_valid_cols(set_names(list(x), name))```
4

0 回答 0