1. Cleaned up a whole lot of logging messages.

2. Start of ability to attach formatters.
3. Better logging of execution time.

Author: dames

column_transformers/StringTransformers.py

@@ -0,0 +1,9 @@
+class ToUpper:
+    def execute(text_in):
+        x = text_in.upper()
+        return x
+
+class TrimWhiteSpace:
+    def execute(text_in):
+        return text_in.strip()
+

column_transformers/ToUpper.py

@@ -1,6 +1,7 @@
 {
+
   "source_table": {
-    "name": "SmallTable",
+    "name": "SourceData",
     "schema": "dbo"
   },
   "stage_table": {
@@ -12,7 +13,7 @@
     "schema": "load"
   },
   "batch": {
-    "size": 1000000,
+    "size": 100000,
     "source_unique_column": "id"
   },
   "columns": [
@@ -41,7 +42,7 @@
         "type": "citext.CIText",
         "nullable": true
       },
-      "column_transformer": "ToUpper.ToUpper"
+      "column_transformer": "StringTransformers.ToUpper"
     },
     {
       "source_name": "IntColumn1",

configuraton/provider.json

@@ -3,7 +3,7 @@
 from io import StringIO
 import importlib
 
-
+from column_transformers.StringTransformers import ToUpper
 
 
 
@@ -33,33 +33,33 @@ def import_batch(self, source_engine, target_engine, target_table_configuration,
         self.logger.debug("ImportBatch Starting for source {0} target {1} previous_key {2}".format(self.source_table_configuration['name'],
                                                                                                    target_table_configuration['name'],
                                                                                                    previous_key))
+
         sql = self.build_select_statement(previous_key)
-        self.logger.debug("SQL Statement: {0}".format(sql))
-        self.logger.info("Starting read")
+
+        self.logger.debug("Starting read of SQL Statement: {0}".format(sql))
         data_frame = pandas.read_sql_query(sql, source_engine)
-        self.logger.info("Completed read")
+        self.logger.debug("Completed read")
 
         batch_tracker.extract_completed_successfully(len(data_frame))
 
-        self.attach_column_transformers(data_frame)
-
         if len(data_frame) == 0:
             self.logger.info("There are no rows to import, returning False")
             batch_tracker.load_skipped_due_to_zero_rows()
             return -1
 
+        data_frame = self.attach_column_transformers(data_frame)
+
         self.write_data_frame_to_table(data_frame, target_table_configuration, target_engine)
         batch_tracker.load_completed_successfully()
 
         last_key_returned = data_frame.iloc[-1][self.batch_configuration['source_unique_column']]
 
-        self.logger.debug("Returning {0} to signify we loaded data.".format(last_key_returned))
-
+        self.logger.info("Batch key {0} Completed. {1}".format(last_key_returned, batch_tracker.get_statistics()))
         return last_key_returned
 
     def write_data_frame_to_table(self, data_frame, table_configuration, target_engine):
         destination_table = "{0}.{1}".format(table_configuration['schema'], table_configuration['name'])
-        self.logger.info("Starting write to table {0}".format(destination_table))
+        self.logger.debug("Starting write to table {0}".format(destination_table))
         data = StringIO()
         data_frame.to_csv(data, header=False, index=False, na_rep='')
         data.seek(0)
@@ -69,32 +69,24 @@ def write_data_frame_to_table(self, data_frame, table_configuration, target_engi
         column_array = list(map(lambda cfg: cfg['destination']['name'], self.columns))
 
         curs.copy_from(data, destination_table, sep=',', columns=column_array, null='')
-        self.logger.info("Completed write to table {0}".format(destination_table))
+        self.logger.debug("Completed write to table {0}".format(destination_table))
 
         curs.connection.commit()
         return
 
     def attach_column_transformers(self, data_frame):
-        return
-        #for column in self.columns:
-            #if 'column_transformer' in column:
-
-                #TODO: this is horribly broken
-                #data_frame = data_frame[column['source_name']].map(lambda x: x.upper())
-                #print (data_frame)
+        self.logger.debug("Attaching column transformers")
+        for column in self.columns:
+            if 'column_transformer' in column:
                 #transformer = self.create_column_transformer_type(column['column_transformer'])
-               #// df['a'] = df['a'].map(lambda a: a / 2.)
-
-                #data_frame.
-
-
+                transformer = ToUpper.execute;
+                data_frame[column['source_name']] = data_frame[column['source_name']].map(transformer)
+                #print (data_frame)
+        return data_frame
 
 
     def create_column_transformer_type(self, type_name):
         module = importlib.import_module(type_name)
         class_ = getattr(module, type_name)
         instance = class_()
         return instance
-
-    def remove_non_existent_columns(self, columns):
-        pass

relational_data_loader_project/BatchDataLoader.py

@@ -35,7 +35,6 @@ def start_single_import(self, source_engine, target_engine, configuration_name,
                                                    columns, target_engine, drop_first=True)
 
         # Import the data.
-        self.logger.info("Creating Batch Importer")
         batch_importer = BatchDataLoader(pipeline_configuration['source_table'], columns,
                                          pipeline_configuration['batch'])
 

relational_data_loader_project/DataLoadManager.py

@@ -2,7 +2,7 @@
 
 
 class DataLoadTracker:
-    started = datetime.now()
+    started = None
     completed = None
     status = "Not Started"
     total_row_count = 0
@@ -18,6 +18,8 @@ def __init__(self, configuration_name, configuration, is_full_load):
         self.configuration_name = configuration_name
         self.configuration = configuration
         self.is_full_load = is_full_load
+        self.started = datetime.now()
+        self.status = "Not Started"
 
     def start_batch(self):
         batch = self.Batch()
@@ -34,29 +36,52 @@ def completed_successfully(self):
         self.rows_per_second = self.total_row_count / self.total_execution_time.total_seconds()
 
     def get_statistics(self):
-        return "Rows: {0}, Total Execution Time: {1}. ({2} rows per second)".format(self.total_row_count,
-                                                                                    self.total_execution_time,
-                                                                                    self.rows_per_second)
+        return "Rows: {0}, Total Execution Time: {1}. ({2:.2f} rows per second)".format(self.total_row_count,
+                                                                                            self.total_execution_time,
+                                                                                            self.rows_per_second)
 
     class Batch:
         row_count = 0
-        extract_started = datetime.now()
-        extract_completed_on = None
-        load_completed_on = None
+        extract_started = None
+        extract_completed = None
+        load_completed = None
         status = "Not Started"
 
+        extract_execution_time = None
+        extract_rows_per_second = 0
+        load_execution_time = None
+        load_rows_per_second = 0
+        total_rows_per_second = 0
+        total_execution_time = None
+
         def __init__(self):
-            pass
+            self.extract_started = datetime.now()
+            self.status = "Not Started"
 
         def extract_completed_successfully(self, row_count):
             self.status = "Extract Completed Successfully"
             self.row_count = row_count
-            self.extract_completed_on = datetime.now()
+            self.extract_completed = datetime.now()
+            self.extract_execution_time = self.extract_completed - self.extract_started
+            self.extract_rows_per_second = self.row_count / self.extract_execution_time.total_seconds()
 
         def load_completed_successfully(self):
             self.status = "Load Completed Successfully"
-            self.load_completed_on = datetime.now()
+            self.load_completed = datetime.now()
+            self.load_execution_time = self.load_completed - self.extract_completed
+            self.load_rows_per_second = self.row_count / self.load_execution_time.total_seconds()
+            self.total_execution_time = self.load_completed - self.extract_started
+            self.total_rows_per_second = self.row_count / self.total_execution_time.total_seconds()
 
         def load_skipped_due_to_zero_rows(self):
             self.status = "Skipped - Zero Rows"
-            self.load_completed_on = datetime.now()
+            self.load_completed = datetime.now()
+
+        def get_statistics(self):
+            return "Rows: {0}, Extract Execution Time: {1} ({2:.2f} rows per second). Load Execution Time {3} ({4:.2f} rows per second) Total Execution Time {5} ({6:.2f} rows per second)".format(self.row_count,
+                                                                                       self.extract_execution_time,
+                                                                                       self.extract_rows_per_second,
+                                                                                       self.load_execution_time,
+                                                                                       self.load_rows_per_second,
+                                                                                       self.total_execution_time,
+                                                                                       self.total_rows_per_second)

relational_data_loader_project/DataLoadTracker.py

@@ -19,7 +19,7 @@ def configure_logging():
     console_stream_handler = logging.StreamHandler()
     console_stream_handler.setFormatter(formatter)
     log.addHandler(console_stream_handler)
-    log.setLevel(logging.DEBUG)
+    log.setLevel(logging.INFO)
     return
 
 def get_arguments():