claims_db/db_loader/mcaid/master_mcaid_analytic.R
In PHSKC-APDE/Medicaid: Query and analyze WA State Medicaid, Medicare, and APCD data
#### MASTER CODE TO UPDATE MEDICAID ANALYTIC TABLES
#
# Alastair Matheson, PHSKC (APDE)
#
# 2020-01
#### Set up global parameter and call in libraries ####
options(max.print = 350, tibble.print_max = 50, warning.length = 8170,
scipen = 999)
library(tidyverse) # Manipulate data
library(data.table) # Manipulate data
library(lubridate) # Manipulate dates
library(odbc) # Read to and write from SQL
library(RCurl) # Read files from Github
library(configr) # Read in YAML files
library(glue) # Safely combine SQL code
db_claims <- dbConnect(odbc(), "PHClaims51")
#### SET UP FUNCTIONS ####
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/add_index.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/alter_schema.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/claim_ccw.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/create_table.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/etl_log.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/load_table.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/qa_load_sql.R")
#### CREATE ELIG TABLES --------------------------------------------------------
#### MCAID_ELIG_DEMO ####
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_elig_demo.R")
# Pull out run date of stage.mcaid_elig_demo
last_run_elig_demo <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_elig_demo")[[1]])
# QA stage version
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_elig_demo.R")
qa_stage_mcaid_elig_demo <- qa_mcaid_elig_demo_f(conn = db_claims, load_only = F)
# Check that things passed QA before loading final table
if (qa_stage_mcaid_elig_demo == 0) {
# Load final table (assumes no changes to table structure)
load_table_from_sql_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_demo.yaml",
truncate = T, truncate_date = F)
# QA final table
qa_rows_final_elig_demo <- qa_sql_row_count_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_demo.yaml",
overall = T, ind_yr = F)
odbc::dbGetQuery(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_elig_demo},
'final.mcaid_elig_demo',
'Number final rows compared to stage',
{qa_rows_final_elig_demo$qa_result},
{Sys.time()},
{qa_rows_final_elig_demo$note})",
.con = db_claims))
rm(qa_rows_final_elig_demo)
} else {
stop("Something went wrong with the mcaid_elig_demo run. See metadata.qa_mcaid_values")
}
rm(qa_stage_mcaid_elig_demo, last_run_elig_demo)
#### MCAID_ELIG_TIMEVAR ####
# Create and load stage version
create_table_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/create_stage.mcaid_elig_timevar.yaml",
overall = T, ind_yr = F, overwrite = T)
system.time(devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_elig_timevar.R"))
# Pull out run date of stage.mcaid_elig_timevar
last_run_elig_timevar <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_elig_timevar")[[1]])
# QA stage version
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_elig_timevar.R")
qa_stage_mcaid_elig_timevar <- qa_mcaid_elig_timevar_f(conn = db_claims, load_only = F)
# Check that things passed QA before loading final table
if (qa_stage_mcaid_elig_timevar == 0) {
# Create and load final table
create_table_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_timevar.yaml",
overall = T, ind_yr = F, overwrite = T)
load_table_from_sql_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_timevar.yaml",
truncate = T, truncate_date = F)
# QA final table
qa_rows_final_elig_timevar <- qa_sql_row_count_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_timevar.yaml",
overall = T, ind_yr = F)
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_elig_timevar},
'final.mcaid_elig_timevar',
'Number final rows compared to stage',
{qa_rows_final_elig_timevar$qa_result},
{Sys.time()},
{qa_rows_final_elig_timevar$note})",
.con = db_claims))
rm(qa_rows_final_elig_timevar)
} else {
stop("Something went wrong with the mcaid_elig_timevar run. See metadata.qa_mcaid_values")
}
rm(last_run_elig_timevar, qa_stage_mcaid_elig_timevar)
#### CREATE CLAIMS TABLES ------------------------------------------------------
# Need to follow this order when making tables because of dependencies
# These scripts depend only on [stage].[mcaid_claim]:
# mcaid_claim_line
# mcaid_claim_icdcm_header
# mcaid_claim_procedure
# mcaid_claim_pharm
#
# The mcaid_claim_header table relies on the tables above
# The mcaid_claim_value_set table relies on the tables above
#### MCAID_CLAIM_LINE ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_line.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_line.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_line failed. See metadata.qa_mcaid for details")
claim_line_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_line passed, loading to final schema")
claim_line_fail <- 0
# Pull out run date of stage.mcaid_claim_line
last_run_claim_line <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_line")[[1]])
# Pull in config files
final_mcaid_claim_line_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_line.yaml"))
archive_mcaid_claim_line_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_line.yaml"))
# Track how many rows in stage
rows_claim_line_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_line"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_line_config$from_schema,
to_schema = archive_mcaid_claim_line_config$to_schema,
table_name = archive_mcaid_claim_line_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_line_config$index_name)
# Alter schema to final table
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_line_config$from_schema,
to_schema = final_mcaid_claim_line_config$to_schema,
table_name = final_mcaid_claim_line_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_line_config$index_name)
# Rename index
DBI::dbExecute(db_claims,
glue::glue_sql("EXEC sp_rename N'final.mcaid_claim_line.{`stage_mcaid_claim_line_config$index_name`}',
N'{`final_mcaid_claim_line_config$index_name`}', N'INDEX';",
.con = db_claims))
# QA final table
rows_claim_line_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_line"))
if (rows_claim_line_stage == rows_claim_line_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_line},
'final.mcaid_claim_line',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_line},
'final.mcaid_claim_line',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_line_final} rows in final schema (expecting {rows_claim_line_stage})')",
.con = db_claims))
}
rm(last_run_claim_line, final_mcaid_claim_line_config, archive_mcaid_claim_line_config,
rows_claim_line_stage, rows_claim_line_final)
}
rm(fail_tot)
#### MCAID_CLAIM_ICDCM_HEADER ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_icdcm_header.R")
### QA table
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_icdcm_header.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_icdcm_header failed. See metadata.qa_mcaid for details")
claim_icdcm_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_icdcm_header passed")
claim_icdcm_fail <- 0
# Pull out run date of stage.mcaid_claim_icdcm_header
last_run_claim_icdcm_header <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_icdcm_header")[[1]])
# Pull in config files
final_mcaid_claim_icdcm_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_icdcm_header.yaml"))
archive_mcaid_claim_icdcm_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_icdcm_header.yaml"))
# Track how many rows in stage
rows_claim_icdcm_header_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_icdcm_header"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_icdcm_header_config$from_schema,
to_schema = archive_mcaid_claim_icdcm_header_config$to_schema,
table_name = archive_mcaid_claim_icdcm_header_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_icdcm_header_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_icdcm_header_config$from_schema,
to_schema = final_mcaid_claim_icdcm_header_config$to_schema,
table_name = final_mcaid_claim_icdcm_header_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_icdcm_header_config$index_name)
# QA final table
rows_claim_icdcm_header_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_icdcm_header"))
if (rows_claim_icdcm_header_stage == rows_claim_icdcm_header_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_icdcm_header},
'final.mcaid_claim_icdcm_header',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_icdcm_header},
'final.mcaid_claim_icdcm_header',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_icdcm_header_final} rows in final schema (expecting {rows_claim_icdcm_header_stage})')",
.con = db_claims))
}
rm(last_run_claim_icdcm_header, final_mcaid_claim_icdcm_header_config,
archive_mcaid_claim_icdcm_header_config, rows_claim_icdcm_header_stage, rows_claim_icdcm_header_final)
}
rm(fail_tot)
#### MCAID_CLAIM_PROCEDURE ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_procedure.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_procedure.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_procedure failed. See metadata.qa_mcaid for details")
claim_procedure_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_procedure passed, loading to final schema")
claim_procedure_fail <- 0
# Pull out run date of stage.mcaid_claim_procedure
last_run_claim_procedure <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_procedure")[[1]])
# Pull in config files
final_mcaid_claim_procedure_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_procedure.yaml"))
archive_mcaid_claim_procedure_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_procedure.yaml"))
# Track how many rows in stage
rows_claim_procedure_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_procedure"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_procedure_config$from_schema,
to_schema = archive_mcaid_claim_procedure_config$to_schema,
table_name = archive_mcaid_claim_procedure_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_procedure_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_procedure_config$from_schema,
to_schema = final_mcaid_claim_procedure_config$to_schema,
table_name = final_mcaid_claim_procedure_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_procedure_config$index_name)
# QA final table
rows_claim_procedure_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_procedure"))
if (rows_claim_procedure_stage == rows_claim_procedure_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_procedure},
'final.mcaid_claim_procedure',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_procedure},
'final.mcaid_claim_procedure',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_procedure_final} rows in final schema (expecting {rows_claim_procedure_stage})')",
.con = db_claims))
}
rm(last_run_claim_procedure, final_mcaid_claim_procedure_config,
archive_mcaid_claim_procedure_config, rows_claim_procedure_stage, rows_claim_procedure_final)
}
rm(fail_tot)
#### MCAID_CLAIM_PHARM ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_pharm.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_pharm.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_pharm failed. See metadata.qa_mcaid for details")
claim_pharm_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_pharm passed, loading to final schema")
claim_pharm_fail <- 0
# Pull out run date of stage.mcaid_claim_pharm
last_run_claim_pharm <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_pharm")[[1]])
# Pull in config files
final_mcaid_claim_pharm_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_pharm.yaml"))
archive_mcaid_claim_pharm_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_pharm.yaml"))
# Track how many rows in stage
rows_claim_pharm_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_pharm"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_pharm_config$from_schema,
to_schema = archive_mcaid_claim_pharm_config$to_schema,
table_name = archive_mcaid_claim_pharm_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_pharm_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_pharm_config$from_schema,
to_schema = final_mcaid_claim_pharm_config$to_schema,
table_name = final_mcaid_claim_pharm_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_pharm_config$index_name)
# QA final table
rows_claim_pharm_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_pharm"))
if (rows_claim_pharm_stage == rows_claim_pharm_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_pharm},
'final.mcaid_claim_pharm',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_pharm},
'final.mcaid_claim_pharm',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_pharm_final} rows in final schema (expecting {rows_claim_pharm_stage})')",
.con = db_claims))
}
rm(last_run_claim_pharm, final_mcaid_claim_pharm_config, archive_mcaid_claim_pharm_config,
rows_claim_pharm_stage, rows_claim_pharm_final)
}
rm(fail_tot)
#### MCAID_CLAIM_HEADER ####
if (sum(claim_line_fail, claim_icdcm_fail, claim_procedure_fail, claim_pharm_fail) > 0) {
stop("One or more claims analytic tables failed, mcaid_claim_header not created. See metadata.mcaid_qa for details")
} else {
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_header.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_header.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_header failed. See metadata.qa_mcaid for details")
} else {
message("All QA checks on stage.mcaid_claim_header passed, loading to final schema")
# Pull out run date of stage.mcaid_claim_header
last_run_claim_header <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_header")[[1]])
# Pull in config files
final_mcaid_claim_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_header.yaml"))
archive_mcaid_claim_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_header.yaml"))
# Track how many rows in stage
rows_claim_header_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_header"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_header_config$from_schema,
to_schema = archive_mcaid_claim_header_config$to_schema,
table_name = archive_mcaid_claim_header_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_header_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_header_config$from_schema,
to_schema = final_mcaid_claim_header_config$to_schema,
table_name = final_mcaid_claim_header_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_header_config$index_name)
# QA final table
rows_claim_header_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_header"))
if (rows_claim_header_stage == rows_claim_header_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_header},
'final.mcaid_claim_header',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_header},
'final.mcaid_claim_header',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_header_final} rows in final schema (expecting {rows_claim_header_stage})')",
.con = db_claims))
}
rm(last_run_claim_header, final_mcaid_claim_header_config, archive_mcaid_claim_header_config,
rows_claim_header_stage, rows_claim_header_final)
}
rm(claim_line_fail, claim_icdcm_fail, claim_procedure_fail, claim_pharm_fail)
rm(fail_tot)
}
#### MCAID_CLAIM_VALUE_SET ####
### CCW
# Load table to SQL
load_ccw(conn = db_claims, source = "mcaid")
# QA table
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_ccw.R",
echo = T)
# If QA passes, load to final table
if (ccw_qa_result == "PASS") {
create_table_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_ccw.yaml",
overall = T, ind_yr = F, overwrite = T)
load_table_from_sql_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_ccw.yaml",
truncate = T, truncate_date = F)
# QA final table
qa_rows_final_claim_ccw <- qa_sql_row_count_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_ccw.yaml",
overall = T, ind_yr = F)
odbc::dbGetQuery(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_ccw},
'final.mcaid_claim_ccw',
'Number final rows compared to stage',
{qa_rows_final_claim_ccw$qa_result},
{Sys.time()},
{qa_rows_final_claim_ccw$note})",
.con = db_claims))
rm(qa_rows_final_claim_ccw)
} else {
warning("CCW table failed QA and was not loaded to final schema")
}
#### PERFORMANCE MEASURES ------------------------------------------------------
# All these tables will be renamed to have a mcaid_ prefix at some point
#### PERF_ELIG_MEMBER_MONTH ####
DBI::dbExecute(db_claims, "EXEC [stage].[sp_perf_elig_member_month]")
#### PERF_ENROLL_DENOM ####
# Need to find which months have been refreshed and run for those
# Assumes a 12-month refresh. Change manually for other options
max_elig_month <- odbc::dbGetQuery(db_claims, "SELECT MAX (CLNDR_YEAR_MNTH) FROM stage.mcaid_elig")[[1]]
min_elig_month <- max_elig_month - 99
# Run stored procedure
DBI::dbExecute(db_claims,
glue_sql("EXEC [stage].[sp_perf_enroll_denom]
@start_date_int = {min_elig_month}, @end_date_int = {max_elig_month}",
.con = db_claims))
rm(max_elig_month, min_elig_month)
#### PERF_DISTINCT_MEMBER ####
DBI::dbExecute(db_claims, "EXEC [stage].[sp_perf_distinct_member]")
#### ASTHMA MEDICATION RATIO ####
#### DROP TABLES NO LONGER NEEDED ####
PHSKC-APDE/Medicaid documentation built on April 3, 2020, 5:32 p.m.
R Package Documentation
Browse R Packages
We want your feedback!
Note that we can't provide technical support on individual packages. You should contact the package authors for that.
#### MASTER CODE TO UPDATE MEDICAID ANALYTIC TABLES
#
# Alastair Matheson, PHSKC (APDE)
#
# 2020-01
#### Set up global parameter and call in libraries ####
options(max.print = 350, tibble.print_max = 50, warning.length = 8170,
scipen = 999)
library(tidyverse) # Manipulate data
library(data.table) # Manipulate data
library(lubridate) # Manipulate dates
library(odbc) # Read to and write from SQL
library(RCurl) # Read files from Github
library(configr) # Read in YAML files
library(glue) # Safely combine SQL code
db_claims <- dbConnect(odbc(), "PHClaims51")
#### SET UP FUNCTIONS ####
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/add_index.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/alter_schema.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/claim_ccw.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/create_table.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/etl_log.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/load_table.R")
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/db_loader/scripts_general/qa_load_sql.R")
#### CREATE ELIG TABLES --------------------------------------------------------
#### MCAID_ELIG_DEMO ####
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_elig_demo.R")
# Pull out run date of stage.mcaid_elig_demo
last_run_elig_demo <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_elig_demo")[[1]])
# QA stage version
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_elig_demo.R")
qa_stage_mcaid_elig_demo <- qa_mcaid_elig_demo_f(conn = db_claims, load_only = F)
# Check that things passed QA before loading final table
if (qa_stage_mcaid_elig_demo == 0) {
# Load final table (assumes no changes to table structure)
load_table_from_sql_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_demo.yaml",
truncate = T, truncate_date = F)
# QA final table
qa_rows_final_elig_demo <- qa_sql_row_count_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_demo.yaml",
overall = T, ind_yr = F)
odbc::dbGetQuery(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_elig_demo},
'final.mcaid_elig_demo',
'Number final rows compared to stage',
{qa_rows_final_elig_demo$qa_result},
{Sys.time()},
{qa_rows_final_elig_demo$note})",
.con = db_claims))
rm(qa_rows_final_elig_demo)
} else {
stop("Something went wrong with the mcaid_elig_demo run. See metadata.qa_mcaid_values")
}
rm(qa_stage_mcaid_elig_demo, last_run_elig_demo)
#### MCAID_ELIG_TIMEVAR ####
# Create and load stage version
create_table_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/create_stage.mcaid_elig_timevar.yaml",
overall = T, ind_yr = F, overwrite = T)
system.time(devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_elig_timevar.R"))
# Pull out run date of stage.mcaid_elig_timevar
last_run_elig_timevar <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_elig_timevar")[[1]])
# QA stage version
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_elig_timevar.R")
qa_stage_mcaid_elig_timevar <- qa_mcaid_elig_timevar_f(conn = db_claims, load_only = F)
# Check that things passed QA before loading final table
if (qa_stage_mcaid_elig_timevar == 0) {
# Create and load final table
create_table_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_timevar.yaml",
overall = T, ind_yr = F, overwrite = T)
load_table_from_sql_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_timevar.yaml",
truncate = T, truncate_date = F)
# QA final table
qa_rows_final_elig_timevar <- qa_sql_row_count_f(conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_elig_timevar.yaml",
overall = T, ind_yr = F)
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_elig_timevar},
'final.mcaid_elig_timevar',
'Number final rows compared to stage',
{qa_rows_final_elig_timevar$qa_result},
{Sys.time()},
{qa_rows_final_elig_timevar$note})",
.con = db_claims))
rm(qa_rows_final_elig_timevar)
} else {
stop("Something went wrong with the mcaid_elig_timevar run. See metadata.qa_mcaid_values")
}
rm(last_run_elig_timevar, qa_stage_mcaid_elig_timevar)
#### CREATE CLAIMS TABLES ------------------------------------------------------
# Need to follow this order when making tables because of dependencies
# These scripts depend only on [stage].[mcaid_claim]:
# mcaid_claim_line
# mcaid_claim_icdcm_header
# mcaid_claim_procedure
# mcaid_claim_pharm
#
# The mcaid_claim_header table relies on the tables above
# The mcaid_claim_value_set table relies on the tables above
#### MCAID_CLAIM_LINE ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_line.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_line.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_line failed. See metadata.qa_mcaid for details")
claim_line_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_line passed, loading to final schema")
claim_line_fail <- 0
# Pull out run date of stage.mcaid_claim_line
last_run_claim_line <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_line")[[1]])
# Pull in config files
final_mcaid_claim_line_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_line.yaml"))
archive_mcaid_claim_line_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_line.yaml"))
# Track how many rows in stage
rows_claim_line_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_line"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_line_config$from_schema,
to_schema = archive_mcaid_claim_line_config$to_schema,
table_name = archive_mcaid_claim_line_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_line_config$index_name)
# Alter schema to final table
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_line_config$from_schema,
to_schema = final_mcaid_claim_line_config$to_schema,
table_name = final_mcaid_claim_line_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_line_config$index_name)
# Rename index
DBI::dbExecute(db_claims,
glue::glue_sql("EXEC sp_rename N'final.mcaid_claim_line.{`stage_mcaid_claim_line_config$index_name`}',
N'{`final_mcaid_claim_line_config$index_name`}', N'INDEX';",
.con = db_claims))
# QA final table
rows_claim_line_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_line"))
if (rows_claim_line_stage == rows_claim_line_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_line},
'final.mcaid_claim_line',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_line},
'final.mcaid_claim_line',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_line_final} rows in final schema (expecting {rows_claim_line_stage})')",
.con = db_claims))
}
rm(last_run_claim_line, final_mcaid_claim_line_config, archive_mcaid_claim_line_config,
rows_claim_line_stage, rows_claim_line_final)
}
rm(fail_tot)
#### MCAID_CLAIM_ICDCM_HEADER ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_icdcm_header.R")
### QA table
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_icdcm_header.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_icdcm_header failed. See metadata.qa_mcaid for details")
claim_icdcm_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_icdcm_header passed")
claim_icdcm_fail <- 0
# Pull out run date of stage.mcaid_claim_icdcm_header
last_run_claim_icdcm_header <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_icdcm_header")[[1]])
# Pull in config files
final_mcaid_claim_icdcm_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_icdcm_header.yaml"))
archive_mcaid_claim_icdcm_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_icdcm_header.yaml"))
# Track how many rows in stage
rows_claim_icdcm_header_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_icdcm_header"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_icdcm_header_config$from_schema,
to_schema = archive_mcaid_claim_icdcm_header_config$to_schema,
table_name = archive_mcaid_claim_icdcm_header_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_icdcm_header_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_icdcm_header_config$from_schema,
to_schema = final_mcaid_claim_icdcm_header_config$to_schema,
table_name = final_mcaid_claim_icdcm_header_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_icdcm_header_config$index_name)
# QA final table
rows_claim_icdcm_header_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_icdcm_header"))
if (rows_claim_icdcm_header_stage == rows_claim_icdcm_header_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_icdcm_header},
'final.mcaid_claim_icdcm_header',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_icdcm_header},
'final.mcaid_claim_icdcm_header',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_icdcm_header_final} rows in final schema (expecting {rows_claim_icdcm_header_stage})')",
.con = db_claims))
}
rm(last_run_claim_icdcm_header, final_mcaid_claim_icdcm_header_config,
archive_mcaid_claim_icdcm_header_config, rows_claim_icdcm_header_stage, rows_claim_icdcm_header_final)
}
rm(fail_tot)
#### MCAID_CLAIM_PROCEDURE ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_procedure.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_procedure.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_procedure failed. See metadata.qa_mcaid for details")
claim_procedure_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_procedure passed, loading to final schema")
claim_procedure_fail <- 0
# Pull out run date of stage.mcaid_claim_procedure
last_run_claim_procedure <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_procedure")[[1]])
# Pull in config files
final_mcaid_claim_procedure_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_procedure.yaml"))
archive_mcaid_claim_procedure_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_procedure.yaml"))
# Track how many rows in stage
rows_claim_procedure_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_procedure"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_procedure_config$from_schema,
to_schema = archive_mcaid_claim_procedure_config$to_schema,
table_name = archive_mcaid_claim_procedure_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_procedure_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_procedure_config$from_schema,
to_schema = final_mcaid_claim_procedure_config$to_schema,
table_name = final_mcaid_claim_procedure_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_procedure_config$index_name)
# QA final table
rows_claim_procedure_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_procedure"))
if (rows_claim_procedure_stage == rows_claim_procedure_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_procedure},
'final.mcaid_claim_procedure',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_procedure},
'final.mcaid_claim_procedure',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_procedure_final} rows in final schema (expecting {rows_claim_procedure_stage})')",
.con = db_claims))
}
rm(last_run_claim_procedure, final_mcaid_claim_procedure_config,
archive_mcaid_claim_procedure_config, rows_claim_procedure_stage, rows_claim_procedure_final)
}
rm(fail_tot)
#### MCAID_CLAIM_PHARM ####
### Create and load table, add index
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_pharm.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_pharm.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_pharm failed. See metadata.qa_mcaid for details")
claim_pharm_fail <- 1
} else {
message("All QA checks on stage.mcaid_claim_pharm passed, loading to final schema")
claim_pharm_fail <- 0
# Pull out run date of stage.mcaid_claim_pharm
last_run_claim_pharm <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_pharm")[[1]])
# Pull in config files
final_mcaid_claim_pharm_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_pharm.yaml"))
archive_mcaid_claim_pharm_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_pharm.yaml"))
# Track how many rows in stage
rows_claim_pharm_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_pharm"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_pharm_config$from_schema,
to_schema = archive_mcaid_claim_pharm_config$to_schema,
table_name = archive_mcaid_claim_pharm_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_pharm_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_pharm_config$from_schema,
to_schema = final_mcaid_claim_pharm_config$to_schema,
table_name = final_mcaid_claim_pharm_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_pharm_config$index_name)
# QA final table
rows_claim_pharm_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_pharm"))
if (rows_claim_pharm_stage == rows_claim_pharm_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_pharm},
'final.mcaid_claim_pharm',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_pharm},
'final.mcaid_claim_pharm',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_pharm_final} rows in final schema (expecting {rows_claim_pharm_stage})')",
.con = db_claims))
}
rm(last_run_claim_pharm, final_mcaid_claim_pharm_config, archive_mcaid_claim_pharm_config,
rows_claim_pharm_stage, rows_claim_pharm_final)
}
rm(fail_tot)
#### MCAID_CLAIM_HEADER ####
if (sum(claim_line_fail, claim_icdcm_fail, claim_procedure_fail, claim_pharm_fail) > 0) {
stop("One or more claims analytic tables failed, mcaid_claim_header not created. See metadata.mcaid_qa for details")
} else {
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/load_stage.mcaid_claim_header.R")
### QA table and load to final
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_header.R",
echo = T)
if (fail_tot > 0) {
message("One or more QA checks on stage.mcaid_claim_header failed. See metadata.qa_mcaid for details")
} else {
message("All QA checks on stage.mcaid_claim_header passed, loading to final schema")
# Pull out run date of stage.mcaid_claim_header
last_run_claim_header <- as.POSIXct(odbc::dbGetQuery(db_claims, "SELECT MAX (last_run) FROM stage.mcaid_claim_header")[[1]])
# Pull in config files
final_mcaid_claim_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_header.yaml"))
archive_mcaid_claim_header_config <- yaml::yaml.load(RCurl::getURL(
"https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/archive/tables/load_archive.mcaid_claim_header.yaml"))
# Track how many rows in stage
rows_claim_header_stage <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM stage.mcaid_claim_header"))
# Alter schema from final to archive
alter_schema_f(conn = db_claims,
from_schema = archive_mcaid_claim_header_config$from_schema,
to_schema = archive_mcaid_claim_header_config$to_schema,
table_name = archive_mcaid_claim_header_config$to_table,
rename_index = T,
index_name = archive_mcaid_claim_header_config$index_name)
# Alter schema from stage to final
alter_schema_f(conn = db_claims,
from_schema = final_mcaid_claim_header_config$from_schema,
to_schema = final_mcaid_claim_header_config$to_schema,
table_name = final_mcaid_claim_header_config$to_table,
rename_index = T,
index_name = final_mcaid_claim_header_config$index_name)
# QA final table
rows_claim_header_final <- as.integer(odbc::dbGetQuery(
db_claims, "SELECT COUNT (*) FROM final.mcaid_claim_header"))
if (rows_claim_header_stage == rows_claim_header_final) {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_header},
'final.mcaid_claim_header',
'Number final rows compared to stage',
'PASS',
{Sys.time()},
'All rows transferred to final schema')",
.con = db_claims))
} else {
DBI::dbExecute(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_header},
'final.mcaid_claim_header',
'Number final rows compared to stage',
'FAIL',
{Sys.time()},
'{rows_claim_header_final} rows in final schema (expecting {rows_claim_header_stage})')",
.con = db_claims))
}
rm(last_run_claim_header, final_mcaid_claim_header_config, archive_mcaid_claim_header_config,
rows_claim_header_stage, rows_claim_header_final)
}
rm(claim_line_fail, claim_icdcm_fail, claim_procedure_fail, claim_pharm_fail)
rm(fail_tot)
}
#### MCAID_CLAIM_VALUE_SET ####
### CCW
# Load table to SQL
load_ccw(conn = db_claims, source = "mcaid")
# QA table
devtools::source_url("https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/stage/tables/qa_stage.mcaid_claim_ccw.R",
echo = T)
# If QA passes, load to final table
if (ccw_qa_result == "PASS") {
create_table_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_ccw.yaml",
overall = T, ind_yr = F, overwrite = T)
load_table_from_sql_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_ccw.yaml",
truncate = T, truncate_date = F)
# QA final table
qa_rows_final_claim_ccw <- qa_sql_row_count_f(
conn = db_claims,
config_url = "https://raw.githubusercontent.com/PHSKC-APDE/claims_data/master/claims_db/phclaims/final/tables/load_final.mcaid_claim_ccw.yaml",
overall = T, ind_yr = F)
odbc::dbGetQuery(
conn = db_claims,
glue::glue_sql("INSERT INTO metadata.qa_mcaid
(last_run, table_name, qa_item, qa_result, qa_date, note)
VALUES ({last_run_claim_ccw},
'final.mcaid_claim_ccw',
'Number final rows compared to stage',
{qa_rows_final_claim_ccw$qa_result},
{Sys.time()},
{qa_rows_final_claim_ccw$note})",
.con = db_claims))
rm(qa_rows_final_claim_ccw)
} else {
warning("CCW table failed QA and was not loaded to final schema")
}
#### PERFORMANCE MEASURES ------------------------------------------------------
# All these tables will be renamed to have a mcaid_ prefix at some point
#### PERF_ELIG_MEMBER_MONTH ####
DBI::dbExecute(db_claims, "EXEC [stage].[sp_perf_elig_member_month]")
#### PERF_ENROLL_DENOM ####
# Need to find which months have been refreshed and run for those
# Assumes a 12-month refresh. Change manually for other options
max_elig_month <- odbc::dbGetQuery(db_claims, "SELECT MAX (CLNDR_YEAR_MNTH) FROM stage.mcaid_elig")[[1]]
min_elig_month <- max_elig_month - 99
# Run stored procedure
DBI::dbExecute(db_claims,
glue_sql("EXEC [stage].[sp_perf_enroll_denom]
@start_date_int = {min_elig_month}, @end_date_int = {max_elig_month}",
.con = db_claims))
rm(max_elig_month, min_elig_month)
#### PERF_DISTINCT_MEMBER ####
DBI::dbExecute(db_claims, "EXEC [stage].[sp_perf_distinct_member]")
#### ASTHMA MEDICATION RATIO ####
#### DROP TABLES NO LONGER NEEDED ####
R Package Documentation
Browse R Packages
We want your feedback!
Note that we can't provide technical support on individual packages. You should contact the package authors for that.
Embedding an R snippet on your website
Add the following code to your website.
For more information on customizing the embed code, read Embedding Snippets.