pvec: O laço perfeito

devtools::install_github("abjur/abjutils")

Como funciona

O pvec() recebe duas informações de entrada: uma lista ou vetor de inputs e uma função a ser aplicada. O pvec() funciona exatamente como um purrr::map(), mas retorna um data.frame com os outputs.

Por exemplo, digamos que nosso objetivo seja aplicar a função

funcao <- function(x) {
  # dorme 1s
  Sys.sleep(1)
  # aplica o log
  log(x)
}

em uma lista de entradas, dada por

input <- list(1, 2, -1, "a")

O resultado é dado por:

resultado <- abjutils::pvec(input, funcao)
resultado

# A tibble: 4 x 3
     id return output           
  <int> <chr>  <list>           
1     1 result <dbl [1]>        
2     2 result <dbl [1]>        
3     3 result <dbl [1]>        
4     4 error  <S3: simpleError>

Ou seja, o resultado é um data.frame, que tem o número de linhas exatamente igual ao comprimento do vetor ou lista de entrada, e três colunas específicas.

1. id, que guarda o índice de entrada. Se a lista de entrada é nomeada, id guarda esses nomes.
2. return identifica se a aplicação retornou num resultado (result) ou erro (error)
3. output é uma coluna-lista que contém os resultados. Quando o resultado é um erro, o erro é capturado e colocado no elemento correspondente.

Ou seja, uma característica do pvec() é que ele nunca irá travar. Se essa operação travar, é porque o computador todo travou.

É importante notar que alguns resultados nesse caso são NaN. Isso ocorre pois log(-1) resulta em NaN, acompanhado de um warning. O pvec() não trabalha com warnings.

Outra característica importante do pvec() é que ele roda em paralelo. Você pode controlar a quantidade de núcleos de processamento com o parâmetro .cores. Por padrão, ele usará o número de núcleos da sua máquina.

Finalmente, o que não poderia faltar no pvec() é a utilização de barras de progresso. Por exemplo, considerando como input

input <- list(a = 1, b = 2, c = -1, d = "a",
              e = 2, f = 3, g = -2, h = "b")

O resultado é

abjutils::pvec(input, funcao)

Progress: ───────────────────────────────                              100%

Progress: ──────────────────────────────────────────────────────────── 100%

# A tibble: 8 x 3
  id    return output           
  <chr> <chr>  <list>           
1 a     result <dbl [1]>        
2 b     result <dbl [1]>        
3 c     result <dbl [1]>        
4 d     error  <S3: simpleError>
5 e     result <dbl [1]>        
6 f     result <dbl [1]>        
7 g     result <dbl [1]>        
8 h     error  <S3: simpleError>

Se você quiser desligar a barra de progresso, basta adicionar .progress = FALSE.

O parâmetro .flatten

Esse é o parâmetro dos preguiçosos (eu que pedi para o Caio adicionar). Em muitas operações, o resultado que sai no output é uma lista de data.frames ou uma lista de vetores. A opção .flatten faz tidyr::unnest(), empilhando os resultados e colando tudo num vetor ou data.frame.

O único problema é que nesse caso não é possível guardar os erros. Por isso, o pvec() retorna um warning:

abjutils::pvec(input, funcao, .flatten = TRUE)

Progress: ──────────────────────────────────────────────────────────── 100%

# A tibble: 6 x 2
  id     output
  <chr>   <dbl>
1 a       0    
2 b       0.693
3 c     NaN    
4 e       0.693
5 f       1.10 
6 g     NaN    

Warning message:
Since '.flatten = TRUE', a total of 2 errors are being ignored     

Note que o resultado tem 6 linhas, menor que a entrada, que tem 8 elementos. Por isso, use .flatten somente quando você tem certeza do que está fazendo.

Por trás dos panos: o furrr

O pvec() só funciona por conta de dois excelentes pacotes:

• o future, que é um novo paradigma de computação em paralelo no R.
• o furrr, que faz todo o trabalho sujo e implementa a maioria das operações do purrr usando future.

Se quiser estudar esses pacotes e implementar suas próprias soluções, recomendo acessar aqui e aqui. Não incluí detalhes desses pacotes aqui para não sair do foco.

Se quiser adicionar opções do future no pvec(), basta adicioná-las na opção .options. Por padrão, passamos furrr::future_options() nesse argumento.

Wrap-up

1. abjutils::pvec() é um map() que roda em paralelo, tem barras de progresso e trata erros automaticamente.
2. Você pode brincar com as opções .cores, .progress e .flatten para controlar o comportamento do pvec(). Tome muito cuidado com o .flatten, pois ele pode não tratar os erros da forma que você imagina!
3. Estude future e furrr se quiser estender as funcionalidades do pvec().

É isso pessoal. Happy coding ;)

P.S.: Código

Como dito no começo deste post, a função pvec() foi removida do abjutils por causa de alguns problemas nas dependências que dificultavam a sua manutenção em um pacote. O código dela está preservado aqui para que você possa implementá-la por conta própria e ainda é possível acessá-la no arquivo do abjutils.

#' @title Verbose, parallel, and safe map-like
#'
#' @description Using the same argument notation as [purrr::map()], this function
#' iterates over a list of inputs `.x`, applying `.f` to each element. It
#' returns a tibble with the id, whether the function returned an error
#' and the output.
#'
#' @importFrom magrittr %>%
#'
#' @param .x A list or atomic vector
#' @param .f A function, formula, or atomic vector (see [purrr::map()])
#' @param ... Other parameters passed on to `.f`
#' @param .cores Number of cores to use when multiprocessing
#' @param .progress Whether or not to display progress
#' @param .flatten If `TRUE`, the errors are filtered from the output,
#' and the returned object is flattened (a vector, a list, or a tibble)
#' @param .options Options passed on to [furrr::future_map()]
#' ([furrr::future_options()] by default)
#'
#' @seealso [purrr::map()], [furrr::future_map()], [furrr::future_options()]
#'
#' @return A tibble with 3 columns: input, return, and output
#' @export
pvec <- function(.x, .f, ..., .cores = get_cores(), .progress = TRUE, .flatten = FALSE, .options = future_options()) {
  .Deprecated("furrr::future_map")

  # Preserve execution plan
  oplan <- future::plan()
  on.exit(future::plan(oplan), add = TRUE)

  # Set execution plan to multicore
  future::plan(future::multicore, workers = .cores)

  # Capture function side-effects
  .f <- purrr::safely(purrr::as_mapper(.f))

  # Run future map
  out <- furrr::future_map(.x, .f, ..., .progress = .progress, .options = .options)

  # Compact with care
  compact_ <- function(x) {
    if (is.null(x[[1]]) && is.null(x[[2]])) {
      return(list(result = NULL))
    }
    else {
      if (length(x$result) == 0) {
        return(list(result = NULL))
      }
      return(purrr::compact(x))
    }
  }

  # Process output
  pout <- out %>%
    purrr::map(compact_) %>%
    purrr::flatten() %>%
    dplyr::tibble(
      id = purrr::`%||%`(names(.x), seq_along(.x)),
      return = names(.), output = .
    )

  # Flatten results if necessary
  if (.flatten) {
    n_error <- length(pout$return[pout$return == "error"])
    if (n_error > 0) {
      warning(
        "Since '.flatten = TRUE', a total of ", n_error,
        " errors are being ignored",
        call. = FALSE
      )
    }

    pout <- pout %>%
      dplyr::filter(return != "error") %>%
      dplyr::select(-return) %>%
      tidyr::unnest()

    if (ncol(pout) == 1) {
      pout <- dplyr::pull(pout, output)
    }
  }

  return(pout)
}

# Get number of available cores
get_cores <- purrr::partial(future::availableCores, constraints = "multicore")

# Import of future_options()
future_options <- furrr::future_options