Можете ли вы распараллелить маневры панели в R? - PullRequest
Новая
       21

Можете ли вы распараллелить маневры панели в R?

2 голосов
/ 28 мая 2020

В моем сценарии R я использую функции pmdplyr mutate_cascade() и tlag() для изменения моих данных, которые содержат более 3 миллионов записей, поэтому код работает очень медленно, но работает. Чтобы ускорить процесс, я попытался добавить функции параллельной обработки multidplyr. Но выкидывает Ошибка: все столбцы в тибле должны быть векторами. x Столбец . - это объект multidplyr_party_df. Это потому, что невозможно запустить pibble pmdplyr в кластере multidplyr? Я новичок как в 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))```
...