# SNOWFLAKE TODO 2015-07-20: # ------------------------------------------------------------ # accounting_no_RED & analytics_no_red # Should be done using makeQry and include all columns EXCEPT ... # ------------------------------------------------------------ # 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 - Snowflake 20150716.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/") if (FALSE) 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()) ### ----------------------------------------------------- ### ## 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") # run_vaccum <- FALSE skip_to_part <- 0 test_run <- 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 <- 7 # cluster.out <- 7 # run_vaccum <- TRUE # skip_to_part <- 3 test_run <- TRUE UNLOAD_ONLY <- TRUE ## There are still some references to cluster. They might crash the whole program if not found cluster.in <- cluster.out <- 7 wh <- "LOOKER_WH_LARGE" dbname <- "prod" dbname <- "dev_engineering" schema <- "production" setSnowflake(wh = wh, dbname = dbname, schema = schema, verbose=TRUE) ## CREATE THE SCHEMAS in SNOWFLAKE for (sch in unique(c("bi", "production", schema))) if (!sfSchemaExists(schema=sch, wh=wh, dbname=dbname)) sfQry(sprintf("CREATE SCHEMA %s", dbschematbl(dbname=dbname, schema=sch)), wh=wh) 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 - snowflake (20150720).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 { ### ------------------------------------------------------------------------ ## ## Before kicking off: ## ## Check that there are no leftover "new" tables from previous run ## If there are, notify engineer for manual check for (tbl in c("analytics", "accounting")) if (length(qShowTables(namelike=paste0(tbl, "_new%"), schema="bi", rowCount=FALSE, showWarnings=FALSE, justnames=TRUE, cluster=cluster.out, wh=wh, dbname=dbname))) { 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(proj="DeNormalizing", subProj="AUTO_RUN", quiet=TRUE, subl=FALSE, load=FALSE, create=FALSE) stamp <- dateStamp() ## Ensure uniqueness of bucket 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 source("~/git/orch/src/MgmtReport/supportFns/automate new_snowflake.r") ## tbls_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 <- force.accounting || !qTableExists("accounting", schema="bi", cluster=cluster.out) || !qRowCount("accounting", schema="bi", cluster=cluster.out) || !AcctBIandProd_are_synced() if (shouldUpdateToday) tbls_to_update <- c("accounting", tbls_to_update) ## accounting has to always be created after analytics, since the update in accounting queries analytics } ## if we're skipping analytics, then drop this table (usually this is only for testing / debugging / error re-reuns) if (skip.analytics) tbls_to_update <- setdiff(tbls_to_update, "analytics") verboseMsg(TRUE, "Updating the following 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= { ## Source main functions for automation of loading and unloading from buckets source("~/git/orch/src/MgmtReport/supportFns/automate new_snowflake.r") debug(updateDeNormTable) setSnowflake(wh=wh, dbname=dbname, schema=schema) 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 , test_run = test_run , snowflake_inuse=TRUE , 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() "REMOVED FOR NOW 2015-07-20 -- recreate using makeQry" if (FALSE) { #### -------------------------------------------- #### ## Update accounting no red if ("accounting" %in% tbls_to_update) { notifyAndEmail(timeStamp("DONE WITH MGMT UPDATE -- updating the 'no-red' tables", sep="-")) sfQry("TRUNCATE bi.accounting_no_red", cluster=cluster.out, wh=wh, dbname=dbname, verbose=TRUE) sfQry(sqlFileToQry("~/git/orch/src/DeNormalizing/BI_tables/INSERT Accounting_no_RED.sql"), cluster=cluster.out, wh=wh, dbname=dbname, verbose=TRUE) notifyAndEmail(timeStamp("DONE WITH THE FULL MGMT UPDATE (incl no-red)", sep="-")) } else { notifyAndEmail(timeStamp("DONE WITH MGMT UPDATE", sep="-")) } #### -------------------------------------------- #### } cat(" -- I believe these are the buckets .... \n") print(sapply(c("accounting", "analytics"), function(tbl) s3_p("BI_Tables", tbl, stamp, prefix=tbl))) sinkOff() print(sapply(c("accounting", "analytics"), function(tbl) s3_p("BI_Tables", tbl, stamp, prefix=tbl)))