Merge pull request #1819 from cmu-delphi/ndefries/backfill/speed3

[Backfill corrections] Use faster deduplicate fn for QC

Author: krivard

backfill_corrections/delphiBackfillCorrection/NAMESPACE

@@ -30,6 +30,7 @@ importFrom(dplyr,arrange)
 importFrom(dplyr,bind_cols)
 importFrom(dplyr,bind_rows)
 importFrom(dplyr,desc)
+importFrom(dplyr,distinct)
 importFrom(dplyr,everything)
 importFrom(dplyr,filter)
 importFrom(dplyr,full_join)

backfill_corrections/delphiBackfillCorrection/R/main.R

@@ -18,8 +18,6 @@
 run_backfill <- function(df, params,
                          refd_col = "time_value", lag_col = "lag", issued_col = "issue_date",
                          signal_suffixes = c(""), indicator = "", signal = "") {
-  df <- filter(df, lag < params$ref_lag + 30) # a rough filtration to save memory
-
   geo_levels <- params$geo_levels
   if ("state" %in% geo_levels) {
     # If state included, do it last since state processing modifies the
@@ -317,14 +315,12 @@ main <- function(params,
 
     msg_ts("Reading in and combining associated files")
     input_data <- lapply(
-      files_list,
-      function(file) {
-        # refd_col and issued_col read in as strings
-        read_data(file) %>%
-          fips_to_geovalue()
-      }
+      files_list, read_data # refd_col and issued_col read in as strings
     ) %>%
-      bind_rows()
+      bind_rows() %>%
+      fips_to_geovalue() %>%
+       # a rough filter to save memory
+      filter(lag < params$ref_lag + 30)
 
     if (nrow(input_data) == 0) {
       warning("No data available for indicator ", input_group$indicator,

backfill_corrections/delphiBackfillCorrection/R/utils.R

@@ -169,6 +169,8 @@ create_dir_not_exist <- function(path)
 #' @return list of input dataframe augmented with lag column, if it
 #'     didn't already exist, and character vector of one or two value
 #'     column names, depending on requested `value_type`
+#'
+#' @importFrom dplyr distinct across
 validity_checks <- function(df, value_types, num_col, denom_col, signal_suffixes,
                             refd_col = "time_value", lag_col = "lag", issued_col = "issue_date") {
   if (!missing(signal_suffixes) && !is.na(signal_suffixes) && !all(signal_suffixes == "") && !all(is.na(signal_suffixes))) {
@@ -205,13 +207,16 @@ validity_checks <- function(df, value_types, num_col, denom_col, signal_suffixes
   }
 
   # Drop duplicate rows.
-  duplicate_i <- duplicated(df)
-  if (any(duplicate_i)) {
+  raw_df_rows <- nrow(df)
+  df <- distinct(df)
+  new_df_rows <- nrow(df)
+  if (raw_df_rows != new_df_rows) {
     warning("Data contains duplicate rows, dropping")
-    df <- df[!duplicate_i,]
   }
 
-  if (anyDuplicated(df[, c(refd_col, issued_col, "geo_value", "state_id")])) {
+  if (new_df_rows != nrow(
+      distinct(df, across(c(refd_col, issued_col, "geo_value", "state_id")))
+    )) {
     stop("Data contains multiple entries with differing values for at",
          " least one reference date-issue date-location combination")
   }