Optimize FHIR melt

[astruebi]

Ziel ist es FHIR melt zu parallisieren oder noch intern schneller zu machen.

[wendtt]

Lösungsvorschlag (erprobt am DIZ Leipzig):

Ergänzung zu
R-cds2db/cds2db_config.toml:

FHIR_MELTING_MAX_CORES             = 6 # Pro Core, d.h. pro Prozess, werden voraussichtlich ca 8 GB RAM benötigt.
FHIR_MELTING_CHUNK_SIZE            = 10000
FHIR_MELTING_CHUNK_LIMIT           = 20 # VORSICHT: Eine Angabe >0 begrenzt die Anzahl der tatsächlich verarbeiteten
                                        # Chunks, es bleiben also ggf. unverarbeitete übrig, die dann bei nachfolgenden
                                        # Läufen der ToolChain schrittweise abgearbeitet werden.

Ergänzung zu
R-etlutils/etlutils/R/global_variables.R:

utils::globalVariables(c("FHIR_MELTING_MAX_CORES", "FHIR_MELTING_CHUNK_SIZE", "FHIR_MELTING_CHUNK_LIMIT"))

Erweiterung der Funktion in
R-cds2db/cds2db/R/05_Convert_Column_Types.R:

library(parallel)

meltCrackedFHIRData <- function(resource_tables, fhir_table_descriptions) {
  names(fhir_table_descriptions) <- tolower(names(fhir_table_descriptions))
  for (i in seq_along(resource_tables)) {
    resource_name <- names(resource_tables)[i]
    if (isSkipMeltingResource(resource_name)) {
      resource_tables[[i]] <- resource_tables[[i]][0]
      print(paste0("Skipping melting of resource table ", resource_name, " due to debug variable 'DEBUG_SKIP_MELTING_RESOURCES.'"))
      next
    }
    fhir_table_description <- fhir_table_descriptions[[resource_name]]
    if (!is.null(fhir_table_description)) {
      brackets <- fhir_table_description@brackets
      if (isIndexedTable(resource_tables[[i]], brackets)) {
        nrow_before_melt <- nrow(resource_tables[[i]])
        print(paste0("Melt table ", resource_name, " with ", nrow_before_melt, " rows."))
        sep <- fhir_table_description@sep
        time0 <- Sys.time()
        # ACHTUNG:
        # An der dafür vorgesehenen Stelle einlesen der Variablen FHIR_MELTING_MAX_CORES, FHIR_MELTING_CHUNK_SIZE und
        # FHIR_MELTING_CHUNK_LIMIT erledigen,
        #   z. B. mit utils::globalVariables(c("FHIR_MELTING_MAX_CORES", "FHIR_MELTING_CHUNK_SIZE", "FHIR_MELTING_CHUNK_LIMIT"))
        # Pro core, d.h. pro Prozess, werden voraussichtlich ca 8-10 GB RAM verwendet.

        n <- nrow(resource_tables[[i]])
        c <- 0
        doParallelMelting <- TRUE
        if (n <= FHIR_MELTING_CHUNK_SIZE) {
          doParallelMelting <- FALSE
        }
        if (! doParallelMelting) {
          print(paste("Resource table has", n, "<=", FHIR_MELTING_CHUNK_SIZE, "rows; parallel processing is NOT applied."))
          resource_tables[[i]] <- fhircrackr::fhir_melt_all(resource_tables[[i]], brackets, sep, column_name_separator = "/")
        } else {
          fhir_melt_p <- function(resTable) {
            print(paste("Melt chunk of", resource_name, "with", nrow(resTable), "rows"))
            resTable <- fhircrackr::fhir_melt_all(resTable, brackets, sep, column_name_separator = "/")
            return(resTable)
          }

          print(paste("Resource table has ", n, ">", FHIR_MELTING_CHUNK_SIZE, "rows; parallel processing is applied."))
          groups <- cut(seq_len(n), breaks = max(2, floor(1:(n/FHIR_MELTING_CHUNK_SIZE))), labels = FALSE)
          rt <- split(resource_tables[[i]], groups)

          if (FHIR_MELTING_CHUNK_LIMIT > 0) {
            if (length(rt) > FHIR_MELTING_CHUNK_LIMIT) {
              print(paste("!!! Applying FHIR_MELTING_CHUNK_LIMIT, melting only", FHIR_MELTING_CHUNK_LIMIT, "of", length(rt),"chunks."))
              rt = rt[seq(1, FHIR_MELTING_CHUNK_LIMIT, 1)]
            }
          }
          c <- length(rt)

          print(paste("Melt table of", resource_name, "with a total of", n, "rows, divided into", length(rt), "chunks."))
          coresUsed = min(max(1, FHIR_MELTING_MAX_CORES), (parallel::detectCores() - 1), length(rt))
          print(paste("Use", coresUsed, "cores"))
          cl <- parallel::makeCluster(max(1, coresUsed), outfile="")
          rt <- parallel::parLapply(cl, X=rt, fun=fhir_melt_p)
          parallel::stopCluster(cl)
          resource_tables[[i]] <- rbindlist(rt)
        }
        duration <- difftime(Sys.time(), time0, units = 'secs')
        nrow_after_melt <- nrow(resource_tables[[i]])
        if (! doParallelMelting) {
          print(paste("Resource table", resource_name, nrow_before_melt, "rows to", nrow_after_melt, "rows in", duration, "seconds."))
        } else {
          print(paste(c, "chunks of", resource_name, "melted to", nrow_after_melt, "rows in", duration, "seconds."))
        }
      }
    }
  }
  return(resource_tables)
}