Замедление при многократных вызовах в память

Question

Скажи у меня 40 сплошных (DoubleType) переменные, которые я сгруппировал в квартили, используя ft_quantile_discretizer, Идентификация квартилей по всем переменным выполняется очень быстро, так как функция поддерживает выполнение нескольких переменных одновременно.

Далее, я хочу один горячий код для этих переменных с пакетами, но в настоящее время не поддерживается функциональность для одного горячего кода всех этих переменных одним вызовом. Так что я пипец ft_string_indexer, ft_one_hot_encoder, а также sdf_separate_column для каждой из переменных с интервалами по одному, просматривая переменные. Это делает работу. Однако, по мере того, как петля прогрессирует, она значительно замедляется. Я думаю, что не хватает памяти, но не могу понять, как запрограммировать это так, чтобы он выполнялся с одинаковой скоростью по переменным.

Если q_vars такое массив символов имен переменных (скажем, 40 из них) для непрерывных переменных, как я могу кодировать это более эффективным способом искры?

for (v in q_vars) {
   data_sprk_q<-data_sprk_q %>% 
       ft_string_indexer(v,paste0(v,"b"),"keep",string_order_type = "alphabetAsc") %>%
       ft_one_hot_encoder(paste0(v,"b"),paste0(v,"bc")) %>%
       sdf_separate_column(paste0(v,"bc"),into=q_vars_cat_list[[v]]) 
}

Я также попытался выполнить как один массивный конвейер со всеми указанными переменными, но это тоже не решило проблему, поэтому я думаю, что это не имеет никакого отношения к самому циклу.

test_text<-paste0("data_sprk_q<-data_sprk_q %>% ", paste0("ft_string_indexer('",q_vars,"',paste0('",q_vars,"','b'),'keep',string_order_type = 'alphabetAsc') %>% ft_one_hot_encoder(paste0('",q_vars,"','b'),paste0('",q_vars,"','bc')) %>% sdf_separate_column(paste0('",q_vars,"','bc'),into=",q_vars_cat_list,")",collapse=" %>% "))
eval(parse(text=test_text))

Любая помощь будет оценена.

Answer 1

В общем, ожидается некоторое (иногда существенное) замедление с длинным ML Pipeline, что является результатом более сложной, чем линейная сложность оптимизатора Catalyst. Если не разделить процесс на несколько конвейеров и разбить линию между ними (используя контрольные точки и записать данные в постоянное хранилище и загрузить их обратно), в данный момент вы мало что можете с этим поделать.

Однако ваш текущий код добавляет ряд проблем в дополнение к этому:

• Если вы не используете более 10 ведер StringIndexer

  ft_string_indexer(v ,paste0(v, "b"), "keep", string_order_type = "alphabetAsc")
  

  просто дублирует метки, назначенные QuantileDiscretizer, С большим количеством уровней поведение становится еще менее полезным при использовании лексикографического порядка.

• Применение One-Hot-Encoding может вообще не потребоваться (и в худшем случае это может быть вредно), в зависимости от нисходящего процесса и даже с линейными моделями, может быть не обязательно (вы можете утверждать, что назначенные метки действительны) порядковые номера, и запись в виде номинальных значений, и увеличение размерности не является желаемым результатом).

• Однако самой большой проблемой является применение sdf_separate_column, Это

  • Увеличивает стоимость вычисления плана выполнения за счет увеличения количества выражений.
  • Увеличивает объем памяти, необходимый для обработки, преобразовывая разреженные данные в плотные.
  • внутренне sparklyr использования UserDefinedFunction по каждому индексу, эффективно вызывая повторное выделение, декодирование и сборку мусора для одной и той же строки, создавая большую нагрузку на кластер.
  • И последнее, но не менее важное: он отбрасывает метаданные столбцов, широко используемые Spark ML.

  Я бы настоятельно рекомендовал не использовать эту функцию здесь. Исходя из ваших комментариев, похоже, что вы хотите поместить столбцы перед передачей результата в какой-то другой алгоритм - для этого вы можете использовать VectorSlicer,

В целом вы можете переписать свой конвейер как

set.seed(1)

df <- copy_to(sc, tibble(x=rnorm(100), y=runif(100), z=rpois(100, 1)))

input_cols <- colnames(df)
discretized_cols <- paste0(input_cols, "_d")
encoded_cols <- paste0(discretized_cols, "_e") %>% setNames(discretized_cols)

discretizer <- ft_quantile_discretizer(
  sc, input_cols = input_cols, output_cols = discretized_cols, num_buckets = 10
)
encoders <- lapply(
  discretized_cols, 
  function(x) ft_one_hot_encoder(sc, input_col=x, output_col=encoded_cols[x])
)

transformed_df <- do.call(ml_pipeline, c(list(discretizer), encoders)) %>%
  ml_fit(df) %>% 
  ml_transform(df)

и применить ft_vector_slicer при необходимости. Например, чтобы взять значения, соответствующие первому, третьему и шестому сегменту, из x вы можете:

transformed_df %>% 
  ft_vector_slicer(
    input_col="x_d_e", output_col="x_d_e_s", indices=c(0, 2, 5))