In my R script, I'm using the pmdplyr functions mutate_cascade() and tlag() to mutate my data, which contains over 3 million records, so the code is extremely slow but it works. In order to speed things up, I tried adding the parallel processing functions of multidplyr. But it's throwing up Error: All columns in a tibble must be vectors. x Column . is a multidplyr_party_df object. Is that because it's not possible to run the pmdplyr pibble on a multidplyr cluster? I'm new to both pmdplyr and multidplyr, so maybe I'm just doing something wrong?
I get a consolidated import_data dataset with variables uuid, location_id, import_date, customer_name, total_value. Anomaly imports can result in huge spikes in total_value so my code seeks to even out the wildly impossible swings of this value (which is relative to each customer):
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))```
