# stop("DO NOT Auto-Source this file") # TEST OUT THAT YOU ARE GOING TO iTERM, NOT R.app if (FALSE) { setGitBranchToSystem() .g() utilSource() options(prompt.full_screenname=TRUE) setPrompt(full=TRUE) if (FALSE) source("~/git/orch/src/MGMT_REPORT_FULL_AUTO.r") } ## TODO: Warp all of this in one big error-catcher warning("TODO 2015-07-01 -- Add updated Periodview BEFORE GPU") warning("Dont forget about DT.store_uses_royaltydollar from\n ~/git/orch/src/DeNormalizing/store_uses_royaltydollar.r") utilSource() setGitBranchToSystem() ## Capture output to log file sinkfile <- newLogFile(name="Automation_ETL+MGMT_LogFile", dir="~/git/orch/log/MgmtReport/") sinkOn(file=sinkfile, zarchive.old.files=TRUE) ## IF CRASHED, USE: sinkOn(file=getSinkFile(), zarchive.old.files=TRUE) ## RESET All before starting .RESET(verbose.fresh=FALSE) setWidth(251, confirm=FALSE) if (.Pfm == "Darwin") subl(getSinkFile()) ### ----------------------------------------------------- ### library(RPostgreSQL) ## MOVED TO UTILS ## # AcctBIandProd_are_synced <- function(cluster.in=9, cluster.out=9) { # ## This function checks that bi.accounting and production.fact_sales are in sync # ## And returns TRUE/FALSE # ## # ## Useful for deteremining whether or not to run an update on bi.accounting # cat("Checking if production.fact_sales is in sync with bi.accounting... ") # colsToPull <- "storeid" # ## Queries # Q.sales <- makeQry(tbl="fact_sales", schema="production", colsToPull=colsToPull, colsToAgg=c(units="sales", "gross"), minDate="2014-09-01", whereIn=list(storeid=c(1, 7, 187, 286)), dateCol="accountingperiodid") # Q.bi <- makeQry(tbl="accounting", schema="bi", colsToPull=colsToPull, colsToAgg=c(units="units", "gross"), minDate="2014-09-01", whereIn=list(storeid=c(1, 7, 187, 286)), dateCol="accounting_month") # ## Execute Queries # DT.sales <- runQry(Q.sales, cluster=cluster.in, key=colsToPull, verbose=FALSE) # DT.bi <- runQry(Q.bi, cluster=cluster.out, key=colsToPull, verbose=FALSE) # ret <- all(DT.sales == DT.bi) # message(sprintf(" [ %s ]", ifelse(ret, "OK", "Needs Updating"))) # return(ret) # } ## THESE ARE THE DEFAULT FLAGS ## ------ DO NOT CHANGE THESE ------ ## ## ------ DO NOT CHANGE THESE ------ ## { force.accounting <- FALSE force.views <- FALSE skip.analytics <- FALSE skip.accounting <- FALSE skip.GPU <- FALSE skip.views <- FALSE dont.email <- FALSE skip.select.in_Denormalizing <- FALSE fail_if_validate_fails <- TRUE fail_if_GPU_fails <- TRUE unique_bucket <- TRUE ## user FALSE when re-running manually cluster.in <- getOption("db.defaultcluster.in") cluster.out <- getOption("db.defaultcluster.out") skip_to_part <- 0 run_vaccum <- FALSE UNLOAD_ONLY <- FALSE } ## ------ DO NOT CHANGE THESE ------ ## ## ------ DO NOT CHANGE THESE ------ ## ## un-comment as neccessary ## ------ CHANGE HERE ------ ## ## DIFFERENT FOR THIS ONE RUN # force.accounting <- TRUE # force.views <- TRUE # skip.analytics <- TRUE skip.accounting <- TRUE skip.GPU <- TRUE skip.views <- TRUE dont.email <- TRUE # skip.select.in_Denormalizing <- TRUE # fail_if_validate_fails <- TRUE # fail_if_GPU_fails <- FALSE # unique_bucket <- FALSE ## use FALSE when re-running manually # cluster.in <- 7 cluster.in <- 9 cluster.out <- 9 # skip_to_part <- 3 # run_vaccum <- TRUE UNLOAD_ONLY <- TRUE dayOfMonth <- as.numeric(format(Sys.Date(), "%d")) notifyAndEmail(timeStamp("BEGGINING AUTO MGMT UPDATE", sep="-")) ## TODO 1 : parrallelize some of this -- but dont choke the system ## TODO 2 : check for changes in the table OR in the views OR in the GPU, and if none, dont execute ## CONFIRM THAT THERE ARE NO CONTRADICTING FLAGS if (isTRUE(force.accounting) && skip.accounting) stop ("Contradicting flags set to TRUE: force.accounting && skip.accounting") if (force.views && skip.views) stop ("Contradicting flags set to TRUE: force.views && skip.views") #### -------------------------------------------- #### ## update "bi.GPU" if (!skip.GPU) { verboseMsg(TRUE, boxit(" ----------- RUNNING GPU UPDATE ----------- "), minw=85, time=TRUE) tryCatch(expr=source("~/git/orch/src/chartio/Create GPU Table from analytics.r") , error=function(e) notifyIfError(expr=e, subj="ERROR in ETL @ GPU", msg="Creating GPU failed (before running the actual ETL)", fail=fail_if_GPU_fails)) # ret.GPU <- try(source("~/git/orch/src/chartio/Create GPU Table from analytics.r")) # notifyIfError(expr=ret.GPU, subj="ERROR in ETL @ GPU", msg="Creating GPU failed (before running the actual ETL)", fail=TRUE) } #### -------------------------------------------- #### ## recreate views if (!skip.views && (force.views || dayOfMonth %% 6 == 0)) { tryCatch(expr=source("~/git/orch/src/DeNormalizing/BI_tables/Create All Views.r") , error=function(e) notifyIfError(expr=e, subj="ERROR in ETL @ Create Views", msg="Creating Views failed (before running the actual ETL)", fail=TRUE)) # ret.views <- try(source("~/git/orch/src/DeNormalizing/BI_tables/Create All Views.r")) # notifyIfError(expr=ret.views, subj="ERROR in ETL @ Create Views", msg="Creating Views failed (before running the actual ETL)", fail=TRUE) } #### -------------------------------------------- #### ## // BEGIN: Main ETL update { ### ------------ ## Check that there are no leftover "new" tables from previous run for (tbl in c("accounting", "analytics")) if (length(qShowTables(namelike=paste0(tbl, "_new%"), schema="bi", rowCount=FALSE, showWarnings=FALSE, justnames=TRUE, cluster=cluster.out))) { msg <- sprintf("There are '_new' tables for %s.\nThis is likely due to a previous error. Try running:\n qShowTables(name='%1$s', schema='bi')\n", tbl) EmailStatusUpdate(status="NEEDS MANUAL ATTENTION", msg=msg, work=TRUE, sms=TRUE, home=FALSE) warning(msg) } ### ------------ setScience("DeNormalizing", quiet=TRUE, subl=FALSE, load=FALSE, create=FALSE) stamp <- dateStamp() ## Ensure uniqueness while (unique_bucket && any(sapply(c("accounting", "analytics"), function(tbl) s3_bucketExists(s3_p("BI_Tables", tbl, stamp, prefix=tbl))))) stamp <- paste0(stamp, "i") ## Source main functions for automation of loading and unloading from buckets .g() # source("~/git/orch/src/automate new.r") source("~/git/orch/src/MgmtReport/supportFns/automate new.r") ## Initially, tables to update is just analytics. Then deciding if to add accounting based on criteria. ## Use force.accounting when there were changes in the sql or columns, and AcctBIandProd_are_synced(), which just checks total dollars, would not catch it tbls_to_update <- c("analytics") ## Decide if accounting should be updated as well if (!skip.accounting) { shouldUpdateToday <- !qTableExists("accounting", schema="bi", cluster=cluster.out) || !qRowCount("accounting", schema="bi", cluster=cluster.out) || !AcctBIandProd_are_synced() ## || (dayOfMonth > 23 || dayOfMonth < 6) && dayOfMonth %% 3 == 2 if (shouldUpdateToday || force.accounting) tbls_to_update <- c("accounting", tbls_to_update) ## accounting has to always be created after analytics, since the update in accounting queries analytics } if (skip.analytics) tbls_to_update <- setdiff(tbls_to_update, "analytics") ## When pulling from production, do accounting first if (cluster.in %in% c(4,7)) tbls_to_update <- sort(tbls_to_update) verboseMsg(TRUE, "Updating tables ", pasteAND(tbls_to_update), minw=75, time=TRUE) Sys.sleep(2) timeCheck <- emptylist(tbls_to_update) ret <- emptylist(tbls_to_update) timeCheck[["start"]] <- Sys.time() tryCatch(expr= { for (tbl in tbls_to_update) { ret[[tbl]] <- { updateDeNormTable( tbl=tbl , schema="bi" , delim="\t" , manifest=FALSE , gzip=TRUE , UNLOAD_ONLY=UNLOAD_ONLY , cluster.in=cluster.in , cluster.out=cluster.out # , folder=src.p("BI_tables") , projName="DeNormalizing" # for science , add_aggregated = FALSE , dontRename=FALSE , stamp=stamp , skip.select = skip.select.in_Denormalizing # (tbl == "accounting") , skip_to_part = skip_to_part , fail_if_validate_fails = fail_if_validate_fails , verbose=TRUE )} timeCheck[[sprintf("end.%s", tbl)]] <- Sys.time() } }, error = function(e) notifyIfError(expr=e, msg=sprintf("There was an error in running\n updateDeNormTable(%s)", tbl), subj="ERROR in ETL Occurred", fail=TRUE)) } ## // END: Main ETL update #### -------------------------------------------- #### warning ("RICK: Remember to wrap these in TRY()") ## It appears "~/git/orch/src/MgmtReport/GL Automation - Mgmt Report Ingestion.r" might be running twice. ## This is to check .duplicate.sourcing.check <- FALSE .rg() verboseMsg(TRUE, " ----------- RUNNING MgmtReport UPDATE ----------- ", minw=75, time=TRUE) skip_query <- FALSE ret.Mgmt_Auto <- source("~/git/orch/src/MgmtReport/GL Automation - Mgmt Report Ingestion.r") #### -------------------------------------------- #### "FOR NOW - NEVER EMAIL. NOT EVEN BY MISTAKE" # if (!dont.email) { # .rg() # verboseMsg(TRUE, " ----------- EMAILING FORECAST UPDATE ----------- ", minw=75, time=TRUE) # ret.Email_Auto <- source("~/git/orch/src/MgmtReport/GL Auto Latest Forecast Update.r") # } else { # verboseMsg(TRUE, " ----------- DID *NOT* SEND EMAIL ----------- ", minw=75, time=TRUE) # } #### -------------------------------------------- #### saveImageTo() #### -------------------------------------------- #### ## Update accounting no red if ("accounting" %in% tbls_to_update) { notifyAndEmail(timeStamp("DONE WITH MGMT UPDATE -- updating the 'no-red' tables", sep="-")) runQry("TRUNCATE bi.accounting_no_red", cluster=cluster.out, verbose=TRUE) runQry(sqlFileToQry("~/git/orch/src/DeNormalizing/BI_tables/INSERT Accounting_no_RED.sql"), cluster=cluster.out, verbose=TRUE) notifyAndEmail(timeStamp("DONE WITH THE FULL MGMT UPDATE (incl no-red)", sep="-")) } else { notifyAndEmail(timeStamp("DONE WITH MGMT UPDATE", sep="-")) } #### -------------------------------------------- #### "&&& TODO: 2015-02-10" ## There is a way to kick off the vaccum asynchronosly. if (run_vaccum) { verboseMsg(TRUE, boxit("BEGINNING VACUUM")) for (tbl in tbls_to_update) { verboseMsg(TRUE, "For tbl = ", tbl) runQry(sprintf("Vacuuming %s", tbl)) } verboseMsg(TRUE, "Done Vacuuming", tbl) } sinkOff()