Merge branch 'development' into refactor-mh

Author: lizihao-anu

SDL_process/dummy_targets.R

@@ -0,0 +1,74 @@
+################################################################################
+# Name of file -  "dummy_targets.R"
+#
+# Description:
+#       A small target example to run as test for BYOC.
+#
+#       To run the targets pipeline, please use:
+#       targets::tar_make(script = 'dummy_targets.R')
+#
+################################################################################
+
+library(logger)
+library(targets) # main package required
+library(tarchetypes) # support for targets
+library(crew) # support for parallel processing
+library(dplyr)
+# Stage 1 - Setup BYOC_MODE in targets -----------------------------------------
+BYOC_MODE <- Sys.getenv("BYOC_MODE")
+BYOC_MODE <- dplyr::case_when(
+  BYOC_MODE %in% c("TRUE", "T", "true", "True") ~ TRUE,
+  BYOC_MODE %in% c("FALSE", "F", "false", "False") ~ FALSE,
+  TRUE ~ NA
+)
+
+if (BYOC_MODE) {
+  logger::log_info("targets file location on Denodo")
+} else {
+  logger::log_info("targets file location is local")
+}
+
+log_threshold(INFO)
+
+# Phase II - Define functions to be used in the test
+get_data <- function(year) {
+  log_info("Starting the test: Data Generation")
+
+  df <- data.frame(
+    x = 1:10,
+    y = runif(10),
+    year = year
+  )
+
+  log_info("Data Generation complete")
+  return(df)
+}
+
+analyze_data <- function(data) {
+  log_info("Starting Data Analysis")
+
+  res <- mean(data$y)
+
+  log_info("Analysis complete")
+  return(res)
+}
+
+years_to_run <- paste0(20, 17:25)
+
+
+# Stage 2 - Set up targets
+#-------------------------------------------------------------------------------
+list(
+  tar_map(
+    list(year = years_to_run),
+    tar_target(aaraw_data, get_data(year = year)),
+    tar_target(aaverage_value, analyze_data(aaraw_data))
+  ),
+  tar_target(aapipeline_status, {
+    log_info("All targets completed successfully.")
+    return("SUCCESS")
+  })
+)
+#-------------------------------------------------------------------------------
+## End of Targets pipeline ##
+#-------------------------------------------------------------------------------

SDL_process/run_sdl.r

@@ -67,10 +67,34 @@ if (tolower(BYOC_MODE) %in% c("true", "t")) {
   BYOC_MODE <- FALSE
 }
 
+store_path <- dplyr::if_else(
+  BYOC_MODE,
+  "/sdl_byoc/_targets",
+  "/conf/sourcedev/Source_Linkage_File_Updates/_targets"
+)
+
 # run_id <- Sys.getenv("run_id")
 # run_date_time <- Sys.getenv("run_date_time")
 run_date_time <- script_run_time
 
+# Include reporting of last run date of ACADME
+
+if (isFALSE(BYOC_MODE)) {
+  denodo_connect <- createslf::get_denodo_connection(BYOC_MODE = BYOC_MODE)
+}
+dplyr::tbl(
+  denodo_connect,
+  dbplyr::in_schema("sdl", "sdl_byoc_acadme_load_detail")
+) %>%
+  dplyr::collect() %>%
+  # Optional: Format the date to look clean first
+  dplyr::mutate(load_str = format(load_date, "%Y-%m-%d %H:%M:%S")) %>%
+  purrr::pwalk(function(data_mart, load_str, ...) {
+    logger::log_info("{data_mart} loaded at {load_str}")
+  })
+if (isFALSE(BYOC_MODE)) {
+  odbc::dbDisconnect(denodo_connect)
+}
 
 write_to_disk <- TRUE
 
@@ -159,54 +183,56 @@ sg_pub_data <- data.frame(
 # just test one year
 year <- "1920"
 
-# Build BYOC Output File Paths
-byoc_output_files <- get_byoc_output_files(
-  year = year,
-  types = c("homelessness", "maternity", "mh") # using homelessness for test purpose. When development is complete, we change to "types = "byoc_input_files""
-) # can always use any other type for testing also
+# # Build BYOC Output File Paths
+# byoc_output_files <- get_byoc_output_files(
+#   year = year,
+#   types = c("homelessness", "maternity", "mh") # using homelessness for test purpose. When development is complete, we change to "types = "byoc_input_files""
+# ) # can always use any other type for testing also
 
+## targets ----
+targets::tar_make(
+  script = "SDL_process/dummy_targets.R",
+  store = store_path
+)
+logger::log_info("Targets finished.")
 # targets::tar_make()
 
 # test homelessness data only
 ## create homelessness data ----
-logger::log_info("Read and process homelessness data")
-hl1 <- read_extract_homelessness(
-  year,
-  denodo_connect = get_denodo_connection(BYOC_MODE = BYOC_MODE),
-  file_path = get_boxi_extract_path(
-    year = year,
-    type = "homelessness",
-    BYOC_MODE = BYOC_MODE
-  ),
-  BYOC_MODE = BYOC_MODE
-) %>% process_extract_homelessness(
-  year = year,
-  write_to_disk = write_to_disk,
-  la_code_lookup = la_code_lookup,
-  sg_pub_data = sg_pub_data,
-  BYOC_MODE = BYOC_MODE,
-  run_id = run_id,
-  run_date_time = run_date_time
-)
-
-logger::log_info("Read and process maternity data")
-maternity <- read_extract_maternity(
-  year,
-  denodo_connect = get_denodo_connection(BYOC_MODE = BYOC_MODE),
-  file_path = get_boxi_extract_path(
-    year = year,
-    type = "maternity",
-    BYOC_MODE = BYOC_MODE
-  ),
-  BYOC_MODE = BYOC_MODE
-) %>%
-  process_extract_maternity(
-    year = year,
-    write_to_disk = TRUE,
-    BYOC_MODE = BYOC_MODE,
-    run_id = run_id,
-    run_date_time = run_date_time
-  )
+# logger::log_info("Read and process homelessness data")
+# hl1 <- read_extract_homelessness(
+#   year,
+#   denodo_connect = get_denodo_connection(BYOC_MODE = BYOC_MODE),
+#   file_path = get_boxi_extract_path(
+#     year = year,
+#     type = "homelessness",
+#     BYOC_MODE = BYOC_MODE
+#   ),
+#   BYOC_MODE = BYOC_MODE
+# ) %>% process_extract_homelessness(
+#   year = year,
+#   write_to_disk = write_to_disk,
+#   la_code_lookup = la_code_lookup,
+#   sg_pub_data = sg_pub_data,
+#   BYOC_MODE = BYOC_MODE,
+#   run_id = run_id,
+#   run_date_time = run_date_time
+# )
+#
+# logger::log_info("Read and process maternity data")
+# maternity <- read_extract_maternity(
+#   year,
+#   denodo_connect = get_denodo_connection(BYOC_MODE = BYOC_MODE),
+#   file_path = get_boxi_extract_path(year, type = "maternity", BYOC_MODE),
+#   BYOC_MODE = BYOC_MODE
+# ) %>%
+#   process_extract_maternity(
+#     year = year,
+#     write_to_disk = TRUE,
+#     BYOC_MODE = BYOC_MODE,
+#     run_id = run_id,
+#     run_date_time = run_date_time
+#   )
 
 logger::log_info("Read and process mental health data")
 mental_health <- read_extract_mental_health(

_targets.R

@@ -33,7 +33,7 @@ controller <- crew::crew_controller_local(
   name = "my_controller",
   # Specify 6 workers for parallel processing - works with 8CPU, 128GB posit session
   workers = 6,
-  seconds_idle = 3
+  seconds_idle = 30
 )
 
 # Targets options
@@ -48,17 +48,16 @@ tar_option_set(
   # format - default is parquet format
   format = "parquet",
   resources = tar_resources(
-    parquet = tar_resources_parquet(compression = "zstd"),
-    qs = tar_resources_qs(preset = "high")
+    parquet = tar_resources_parquet(compression = "zstd")
   ),
   # error - if an error occurs, the pipeline will continue
   error = "continue",
   # storage - the worker saves/uploads the value.
   storage = "worker",
   # retrieval - the worker loads the target's dependencies.
-  retrieval = "worker",
+  retrieval = "auto",
   # memory - default option: the target stays in memory until the end of the pipeline
-  memory = "persistent",
+  memory = "auto",
   # controller - A controller or controller group object produced by the crew R package
   controller = controller
 )

_targets.yaml

@@ -1,6 +1,6 @@
 main:
-  store: /conf/sourcedev/Source_Linkage_File_Updates/_targets
-  workers: '16'
-  reporter_make: timestamp_positives
+  workers: '8'
+  reporter: verbose
   reporter_outdated: forecast
-  seconds_interval: 30
+  seconds_meta_append: 30
+  store: /sdl_byoc/_targets