Geekly articles weekly

El uso de R para tareas de utilidad

Una buena herramienta + la disponibilidad de habilidades para trabajar con ella, que se logra a través de la práctica, le permite resolver de manera fácil y elegante muchas tareas atípicas "similares". A continuación hay un par de ejemplos similares. Estoy seguro de que muchos pueden ampliar esta lista.


Es una continuación de publicaciones anteriores .


Application Log Analytics


Muy popular es la tarea de realizar cálculos analíticos basados ​​en registros de aplicaciones. Por ejemplo, realice un análisis de las acciones del usuario y calcule indicadores de pronóstico, o pruebe hipótesis. Puede seguir la versión clásica y aumentar la pila ELK o similar (recientemente, Splunk ha abandonado los sistemas disponibles en Rusia). Pero puede pensar un poco y rápidamente hacer todo en R. Rápidamente en todos los sentidos, tanto en la implementación como en el tiempo de procesamiento.


Pero hay una serie de características al resolver un problema similar:


  1. Normalmente, los archivos de registro se escriben en el formato clásico log4j : marca de tiempo, importancia, tipo de subsistema, cuerpo del mensaje.
  2. La marca de tiempo puede contener eventos con una resolución de milisegundos, que deben conservarse para la precisión de los análisis posteriores. Milisegundos pueden escribir sin cumplir con ISO 8601.
  3. El cuerpo del mensaje es una entidad prácticamente desestructurada. Los desarrolladores escriben allí todo lo que consideren necesario, sin limitarse a ningún formato de presentación.
  4. A veces, el cuerpo del mensaje es multilínea, por ejemplo, la salida de la pila de llamadas de Java o el paquete de intercambio intersistema xml. Es necesario reconstruir grabaciones de varias líneas en una (un marcador de marca de tiempo es un signo del comienzo de la grabación).
  5. Un número de atriutes puede ser externo al contenido y deben obtenerse de una manera diferente, por ejemplo, la identificación del objeto puede codificarse en el nombre del archivo de registro.
  6. Los registros en forma de archivos pueden ser de varios megabytes o cientos de gigabytes.
  7. La tarea es muy paralela.

De hecho, la tarea se puede dividir en 2 pasos:


  • preprocesamiento de datos sin procesar;
  • análisis posteriores

El contenido del último paso está determinado por el área temática y las tareas comerciales, R es ideal para este paso. Muchas personas no lo saben, pero el primer paso también se puede resolver con bastante facilidad con R. Además, dependiendo del tamaño de los archivos de registro, se puede agregar un resultado de preprocesamiento parcialmente estructurado adecuado para un análisis posterior a los archivos, así como a la base de datos. Los terabytes muelen uno o dos.


Solo un código de ejemplo:
 library(readr) library(tidyverse) library(magrittr) library(stringi) library(fs) library(glue) library(RClickhouse) library(DBI) library(anytime) library(tictoc) library(iterators) library(foreach) library(doParallel) library(futile.logger) library(re2r) library(data.table) library(future) library(doFuture) common_logname <- "DEV_log_parser.log" table_name <- "DEV_LOGS" flog.appender(appender.file(common_logname)) flog.threshold(INFO) flog.info("Start batch processing") oneTimeProcessing <- function(f_iter, log_type = c("app", "system")) { log_type <- match.arg(log_type) checkmate::assertNames(names(f_iter), permutation.of = c("fname", "short_fname", "location", "wk", "size", "id")) cfg <- list(app = list(db_table = "DEV_APP_LOGS"), system = list(db_table = "DEV_LOGS")) #   data <- readr::read_lines(file = f_iter$fname, progress = FALSE) log_df <- setDT(tibble::enframe(data, name = NULL)) %>% .[, log_line_start := re2r::re2_detect(value, pattern = "^\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}", parallel = F)] %>% .[, log_line_number := cumsum(log_line_start)] %>% .[, body := stri_c(value, collapse = "\n"), by = log_line_number] %>% .[, `:=`(value = NULL, log_line_start = NULL, log_line_number = NULL)] %>% tibble::as_tibble() %>% #  body = character(0)      0  #      POSIXct tidyr::extract(col = "body", into = c("timestamp", "tz", "level", "module", "class", "message"), # tz   (  DEV),      ( DEV) regex = "^(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}:\\d+([+-]\\d+)?) (.*?) <(.*?)> \\[(.*?)\\] (?s:(.*))$", case_insensitive = TRUE, ignore.case = TRUE) %>% #     ISO         (   ?) #  ISO 8601 (https://en.wikipedia.org/wiki/ISO_8601) mutate_at("timestamp", re2r::re2_replace, # tz   (  DEV),      ( DEV) pattern = "(.*) (\\d{2}:\\d{2}:\\d{2}):(\\d+([+-]\\d+)?)", replacement = "\\1T\\2.\\3") %>% mutate_at("timestamp", lubridate::as_datetime, tz = "Europe/Moscow") %>% #    mutate(location = f_iter$location, wk = f_iter$wk) # TRUNCATE  CH    ,           #    CH, ms    (timestamp %% 1) conn <- DBI::dbConnect(RClickhouse::clickhouse(), host = "10.0.0.1", db = "DEV_LOGS") # m <- DBI::dbExecute(conn, glue("ALTER TABLE {table_name}")) write_res <- log_df %>% mutate(ms = (as.numeric(timestamp) %% 1) * 1000) %>% select(location, wk, timestamp, ms, level, module, class, message) %>% #            DBI::dbWriteTable(conn, cfg[[log_type]][["db_table"]], ., append = TRUE) DBI::dbDisconnect(conn) #       res <- tibble::tibble(id = f_iter$id, lines = nrow(log_df), min_t = min(log_df$timestamp), max_t = max(log_df$timestamp), write_res) rm(data, log_df) return(res) } #    tic("Batch processing") #    gc(full = TRUE) nworkers <- parallel::detectCores() - 1 registerDoFuture() # future::plan(multiprocess) # future::plan(multisession) future::plan(multisession, workers = nworkers) # future::plan(sequential) #  ~  #      CH #   ------------------ fnames_tbl <- here::here("raw_data") %>% fs::dir_ls(recurse = TRUE, glob = "*dev_app*.gz") %>% enframe(name = "fname") %>% #         mutate(short_fname = as.character(fs::path_rel(fname, start = "./raw_data"))) %>% select(-value) %>% mutate(size = fs::file_size(fname)) %>% tidyr::extract(col = "short_fname", into = c("location", "wk"), regex = "^([^/]+)/wk(\\d+)", remove = FALSE) %>% arrange(size) %>% mutate(id = paste(format(row_number(), justify = "r", width = 4), "/", n())) %>% #   ~ N  mutate(chunk = (row_number() %% nworkers + 1)) %>% #    ,  dopar    arrange(chunk) start_time <- Sys.time() stat_list <- foreach(it = iter(fnames_tbl, by = "row"), .export = c("start_time"), .verbose = TRUE, .inorder = FALSE, .errorhandling = "remove") %dopar% { #   flog.appender(appender.file(common_logname)) # flog.info(capture.output(gc(verbose = TRUE))) res <- oneTimeProcessing(it, log_type = "app") flog.info(glue("Step {it$id} finished.", "Elapsed {round(difftime(Sys.time(), start_time, units = 'mins'), digits = 2)} min(s) ----------->", .sep = " ")) return(res) } flog.info("Load finished") #    -------------- #    ,    future::plan(sequential) gc(reset = TRUE, full = TRUE) flog.info(capture.output(toc())) #     ------------- logstat_tbl <- stat_list %>% dplyr::bind_rows() %>% #    left_join(fnames_tbl, by = "id") %>% #          mutate(delta_t = as.numeric(difftime(max_t, min_t, units = "mins"))) %>% arrange(min_t) write_delim(logstat_tbl, here::here("output", "DEV_parse_stat.csv.gz"), delim = ";") # ,     ? if(nrow(logstat_tbl) < nrow(fnames_tbl)){ flog.error("!!!!!!! Not all workers were executed successfully !!!!!!!!!") }

Este ejemplo de código contiene conceptos básicos como la paralelización, el tiempo de procesamiento teniendo en cuenta milisegundos, el almacenamiento en la base de datos, la contabilidad de registros de varias líneas, el resumen de los resultados del trabajo, el uso de atributos externos, la evaluación comparativa preliminar y la elección de las funciones y paquetes re2r ( re2r , por ejemplo; esto La biblioteca de Google para trabajar con las regulares es la más rápida y se usa mucho donde, tome la misma ClickHouse mencionada en el código { bencmark , algunos operadores pueden tener ILV cerrado}). Pero el código no pretende ser ideal, ya que es solo una acción única en el preprocesamiento de datos. Lo hace rápida y correctamente, bueno, está bien. Para otra tarea similar, corregimos, teniendo en cuenta resp. datos de entrada


¿Será sorprendentemente más rápido en términos de tiempo obtener el resultado final en otros idiomas? La pregunta está abierta. Las versiones paralelas con python , perl , awk no mostraron diferencias notables. Es posible que el gurú en python logre mejores resultados, pero no olvide que esto es solo una tarea periódica "única".


Restaurando el orden en fotos


Después de un viaje con varios dispositivos a mano, debe recopilar todas las fotos juntas y organizarlas de alguna manera antes de continuar con el procesamiento. Una de las mejores opciones es nombrar los archivos por la fecha de disparo ( YYYY-MM-DD hh_mm_ss ), asegurando así el orden de la foto en la flecha de tiempo. Los atributos Exif ayudan a resolver este problema en un solo paso.


Y esto también se puede hacer usando R en un "par de líneas". exifr y exifr para ayudar.


  • hizo una lista de archivos;
  • sacó atributos;
  • archivos copiados con cambio de nombre acc. con los atributos correctos.

De hecho, la tarea se redujo a la anterior, solo los atributos se recopilan no por el nombre del archivo, sino por sus atributos exif, y en el procesamiento simplemente se copia el archivo con el cambio de nombre. El esqueleto del guión y la lógica del trabajo permanecen sin cambios.


Ejemplo de código de mano rápida:
 library(tidyverse) library(magrittr) library(stringi) library(lubridate) library(stringi) library(fs) library(glue) library(futile.logger) library(anytime) library(tictoc) library(bench) library(exifr) library(tictoc) input_path <- "S:/ " %>% fs::path_real() #       output_path <- "S:/ " %>% fs::path_real() i_fnames <- input_path %>% fs::dir_ls(recurse = TRUE, regexp = "(JPG|jpg)$") raw_df <- read_exif(i_fnames, tags = c("SourceFile", "Model", "DateTimeOriginal")) %>% #      base64, ,    mutate(tmp = sub("^base64:(.*)", "\\1", SourceFile)) %>% mutate(i_fname = purrr::map_chr(tmp, ~rawToChar(jsonlite::base64_dec(.)))) %>% mutate(tm = anytime::anytime(DateTimeOriginal)) %>% select(i_fname, DateTimeOriginal, model = Model, tm) #         clean_df <- raw_df %>% mutate(timestamp = case_when( model == 'iPhone ...' ~ tm, model == 'Nikon ...' ~ tm - lubridate::minutes(56), model == 'Samsung ...' ~ tm - lubridate::minutes(62), TRUE ~ tm) ) %>% mutate_at("i_fname", fs::path_real) %>% mutate(fname = format(timestamp, format = '%Y-%m-%d %H_%M_%S')) %>% #  ,     (""),      mutate(fname = dplyr::coalesce(fname, fs::path_ext_remove(fs::path_file(i_fname)))) %>% #  ,         group_by(fname) %>% mutate(n = n(), idx = row_number()) %>% ungroup() %>% #        mutate(fname = case_when( n > 1 ~ stri_c(fname, '_', idx), TRUE ~ fname ) ) %>% mutate(o_fname = fs::path(!!output_path, paste0(fname, ".jpg"))) #     janitor::get_dupes(clean_df, o_fname) #   tic(" ") clean_df %$% # purrr::walk2(i_fname, o_fname, ~print(glue("{.x} -> {.y}"))) purrr::walk2(i_fname, o_fname, ~fs::file_copy(.x, .y, overwrite = TRUE)) toc() #

¿Por qué exifr ? Porque es un contenedor para la potente herramienta multiplataforma ExifTool .


Tal vez la tarea parezca sintética, lo cual es difícil de discutir, ya que hay muchas utilidades e GUI diferentes para trabajar con Exif y renombrar, pero hay un matiz. No todos los dispositivos pueden seleccionar la zona horaria modificada y ajustar la hora (cámaras, por ejemplo, ¿con qué frecuencia el usuario de la cámara establece la hora exacta en ella?), Por lo que durante el cambio de nombre también debe cambiar las marcas de tiempo en función de la fuente.


Finalización


Hay muchos problemas similares, muchos de ellos pueden resolverse con la ayuda de R también.


Publicación anterior - "Niños, Matemáticas y R" .

Source: https://habr.com/ru/post/464849/

More articles:


All Articles