visualization code for data blog#7 and #17

aggregate_tables.R

@@ -1,6 +1,7 @@
 library(dplyr)
 library(DBI)
 library(rjson)
+library(parallel)
 
 source('s_dplyr.R')
 source('aggregate_tables/indicator_functions.R')
@@ -17,6 +18,11 @@ get_db_connection <- function(system_config_path='config_system.json') {
     return(db)
 }
 
+get_cores <- function(system_config_path='config_system.json') {
+  config <- fromJSON(file=system_config_path)[['data_platform']]
+  if ('parallel' %in% names(config)) {return(config[['parallel']][['cores']])} else {return (1)}
+}
+
 drop_tables <- function(file) {
   config <- fromJSON(file=file)
 
@@ -32,26 +38,38 @@ write_tables <- function(file, debug) {
     db <- get_db_connection()
     for (table.info in config) {
         print(paste('Computing indicators for ', table.info$table, 'indicator table.'))
-        df <- compute_indicators(table.info, db, debug)
+        df <- compute_indicators(table.info, debug)
         print(paste('Writing ', table.info$table, 'indicator table.'))
         dbRemoveTable(db$con, name=table.info$table)
         copy_to(db, df=df, name=table.info$table, temporary=FALSE)
     }
 }
 
-compute_indicators <- function(info, db, debug) {
+compute_indicators <- function(info, debug) {
   debug <- as.logical(debug)
   if (debug == T) {limit = 5000} else {limit = -1}
-    dfs <- lapply(info$components, function(component) {
+  cores <- get_cores()
+  cl <- makeCluster(cores, outfile='/tmp/aggregation.log', type="FORK")
+  clusterExport(cl,varlist=c('info','limit'),envir=environment())
+  clusterExport(cl,varlist=c('get_data_source','s_group_by','aggregate','get_db_connection','fromJSON'))
+
+  dfs <- parLapply(cl,info$components, function(component) {
         print(paste('Getting data source ', component$table))
+        source('s_dplyr.R')
+        source('aggregate_tables/indicator_functions.R')
+        source('data_sources.R')
+        db <- get_db_connection()
         source.data <- get_data_source(db, component$table, limit)
         group.by.str <- paste(info$by, collapse=', ')
         print(paste('Grouping and aggregating', component$table))
         df <- source.data %.% s_group_by(group.by.str) %.% aggregate(component$columns)
+        print(paste('Returning aggregated data from ', component$table))
         return(df)
     })
+
     print('merging...')
     merged <- Reduce(function(...) merge(..., all.x=TRUE, all.y=TRUE, by=info$by), dfs)
+    stopCluster(cl)
     return(merged)
 }