Implimented logging to DataLoadExecution table

Author: davidames

integration_tests/mssql_source/dockerfile

@@ -0,0 +1,3 @@
+FROM microsoft/mssql-server-linux:2017-latest
+
+ENV ACCEPT_EULA=Y

modules/DataLoadManager.py

@@ -1,33 +1,34 @@
 import os
 import json
+import uuid
 import logging
 from modules.BatchDataLoader import BatchDataLoader
 from modules.DestinationTableManager import DestinationTableManager
-from modules.DataLoadTracker import DataLoadTracker
-
+from modules.data_load_tracking.DataLoadTracker import DataLoadTracker
 
 
 class DataLoadManager(object):
-    def __init__(self, configuration_path, data_source, logger=None):
+    def __init__(self, configuration_path, data_source, data_load_tracker_repository, logger=None):
         self.logger = logger or logging.getLogger(__name__)
         self.configuration_path = configuration_path
         self.data_source = data_source
-
+        self.data_load_tracker_repository = data_load_tracker_repository
+        self.correlation_id = uuid.uuid4()
     def start_imports(self, target_engine, full_refresh):
         for file in os.listdir(self.configuration_path):
             self.start_single_import(target_engine, file, full_refresh)
 
         self.logger.info("Execution completed.")
 
-    def start_single_import(self, target_engine, configuration_name, requested_full_refresh):
-        self.logger.debug("Using configuration file : {0}".format(configuration_name))
+    def start_single_import(self, target_engine, model_name, requested_full_refresh):
+        self.logger.debug("Using configuration file : {0}".format(model_name))
 
-        config_file = os.path.abspath(self.configuration_path + configuration_name)
+        config_file = os.path.abspath(self.configuration_path + model_name)
         self.logger.debug("Using configuration file : {0}".format(config_file))
         with open(config_file) as json_data:
             pipeline_configuration = json.load(json_data)
 
-        self.logger.info("Execute Starting for: {0} requested_full_refresh: {1}".format(configuration_name, requested_full_refresh))
+        self.logger.info("Execute Starting for: {0} requested_full_refresh: {1}".format(model_name, requested_full_refresh))
 
         destination_table_manager = DestinationTableManager(target_engine)
 
@@ -41,14 +42,13 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
         self.data_source.assert_data_source_is_valid(pipeline_configuration['source_table'],
                                                      pipeline_configuration['columns'])
 
-        last_sync_version = destination_table_manager.get_last_sync_version(pipeline_configuration['target_schema'],
-                                                                            pipeline_configuration['load_table'])
+        last_sync_version = self.data_load_tracker_repository.get_last_sync_version(model_name)
 
         change_tracking_info = self.data_source.init_change_tracking(pipeline_configuration['source_table'],
                                                                      last_sync_version)
 
 
-        data_load_tracker = DataLoadTracker(configuration_name, json_data, full_refresh, change_tracking_info)
+        data_load_tracker = DataLoadTracker(model_name, json_data, full_refresh, change_tracking_info, self.correlation_id)
 
 
         self.logger.debug(" Change Tracking: this_sync_version: {0} next_sync_version: {1} force_full_load:{2}    : ".format(change_tracking_info.this_sync_version, change_tracking_info.next_sync_version, change_tracking_info.force_full_load()))
@@ -97,5 +97,6 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
             destination_table_manager.drop_table(pipeline_configuration['target_schema'],
                                                  pipeline_configuration['stage_table'])
         data_load_tracker.completed_successfully()
-        self.logger.info("Import Complete for: {0}. {1}".format(configuration_name, data_load_tracker.get_statistics()))
+        self.data_load_tracker_repository.save(data_load_tracker)
+        self.logger.info("Import Complete for: {0}. {1}".format(model_name, data_load_tracker.get_statistics()))
 

modules/DestinationTableManager.py

@@ -12,7 +12,6 @@ class DestinationTableManager(object):
     TIMESTAMP_COLUMN_NAME = "data_pipeline_timestamp"
     IS_DELETED_COLUMN_NAME = "data_pipeline_is_deleted"
     CHANGE_VERSION_COLUMN_NAME = "data_pipeline_change_version"
-    NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME = "data_pipeline_next_change_minimum_version"
 
     def __init__(self, target_engine, logger=None):
         self.logger = logger or logging.getLogger(__name__)
@@ -57,9 +56,6 @@ def create_table(self, schema_name, table_name, columns_configuration, drop_firs
         table.append_column(
             Column(self.CHANGE_VERSION_COLUMN_NAME, BigInteger))
 
-        table.append_column(
-            Column(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME, BigInteger))
-
         if drop_first:
             self.logger.debug(
                 "Dropping table {0}.{1}".format(schema_name, table_name))
@@ -79,16 +75,6 @@ def create_column(self, configuration):
                       primary_key=configuration.get("primary_key", False),
                       nullable=configuration['nullable'])
 
-    # TODO: this should come from a different log table which is only written to at the end of a successful load load.
-    def get_last_sync_version(self, schema_name, table_name):
-        sql = "SELECT max({0}) as version FROM {1}.{2}".format(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME, schema_name, table_name)
-
-        result = self.target_engine.execute(sql)
-        row = result.fetchone()
-        if row["version"] is None:
-            return 0
-        return row["version"]
-
 
     def rename_table(self, schema_name, source_table_name, target_table_name):
 
@@ -135,7 +121,6 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
         column_list = column_list + ",{0}".format(self.TIMESTAMP_COLUMN_NAME)
         column_list = column_list + ",{0}".format(self.IS_DELETED_COLUMN_NAME)
         column_list = column_list + ",{0}".format(self.CHANGE_VERSION_COLUMN_NAME)
-        column_list = column_list + ",{0}".format(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME)
 
 
         primary_key_column_array = [column_configuration['destination']['name'] for column_configuration in columns_configuration if 'primary_key' in column_configuration['destination'] and column_configuration['destination']['primary_key']]
@@ -159,8 +144,6 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
         sql_builder.write(os.linesep)
         sql_builder.write("{0} = EXCLUDED.{0},".format(self.CHANGE_VERSION_COLUMN_NAME))
         sql_builder.write(os.linesep)
-        sql_builder.write("{0} = EXCLUDED.{0}".format(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME))
-        sql_builder.write(os.linesep)
 
         self.logger.debug("Upsert executing {0}".format(sql_builder.getvalue()))
         self.target_engine.execute(sql_builder.getvalue())

modules/RelationalDataLoader.py

@@ -2,7 +2,9 @@
 import argparse
 from sqlalchemy import create_engine
 from modules.DataLoadManager import DataLoadManager
+from modules.data_load_tracking.DataLoadTrackerRepository import DataLoadTrackerRepository
 from modules.data_sources.DataSourceFactory import DataSourceFactory
+from sqlalchemy.orm import sessionmaker
 
 _LOG_LEVEL_STRINGS = ['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG']
 
@@ -20,7 +22,10 @@ def main(self):
 
         destination_engine = create_engine(args['destination-engine'])
 
-        data_load_manager = DataLoadManager(args['configuration-folder'], data_source)
+        session_maker = sessionmaker(bind=destination_engine)
+        repository = DataLoadTrackerRepository(session_maker)
+        repository.create_tables(destination_engine)
+        data_load_manager = DataLoadManager(args['configuration-folder'], data_source, repository)
         data_load_manager.start_imports(destination_engine, args['full_refresh'])
 
     def configure_logging(self, log_level):

modules/batch_executions/BatchExecution.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Column, DateTime, Integer, String, Boolean, BigInteger
+from sqlalchemy.sql import func
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+
+class DataLoadExecution(Base):
+    __tablename__ = 'data_load_execution'
+    __table_args__ = {'schema': 'data_pipeline'}
+    id = Column(Integer, primary_key=True)
+    model_name = Column(String(250), nullable=False)
+    is_full_refresh = Column(Boolean, nullable=False)
+    this_sync_version = Column(BigInteger, nullable=False)
+    next_sync_version = Column(BigInteger, nullable=False)
+    completed_on = Column(DateTime(timezone=True), server_default=func.now())
+    execution_time_ms = Column(Integer, nullable=False)
+    rows_processed = Column(Integer, nullable=False)
+    correlation_id = Column(String(250), nullable=True)
+    status = Column(String(25), nullable=False)
+

modules/batch_executions/BatchExecutionManager.py

@@ -7,20 +7,22 @@ class DataLoadTracker:
     status = "Not Started"
     total_row_count = 0
     batches = []
-    configuration_name = None
+    model_name = None
     configuration = None
-    is_full_load = False
+    is_full_refresh = False
     total_execution_time = None
     total_row_count = 0
     rows_per_second = 0
+    correlation_id = None,
 
-    def __init__(self, configuration_name, configuration, is_full_load, change_tracking_info):
-        self.configuration_name = configuration_name
+    def __init__(self, model_name, configuration, is_full_refresh, change_tracking_info, correlation_id):
+        self.model_name = model_name
         self.configuration = configuration
-        self.is_full_load = is_full_load
+        self.is_full_refresh = is_full_refresh
         self.started = datetime.now()
         self.status = "Not Started"
         self.change_tracking_info = change_tracking_info
+        self.correlation_id = correlation_id
 
     def start_batch(self):
         batch = self.Batch()
@@ -37,7 +39,7 @@ def completed_successfully(self):
         self.rows_per_second = self.total_row_count / self.total_execution_time.total_seconds()
 
     def get_statistics(self):
-        load_type = 'full' if self.is_full_load else "incremental from version {0} ".format(self.change_tracking_info.this_sync_version)
+        load_type = 'full' if self.is_full_refresh else "incremental from version {0} ".format(self.change_tracking_info.this_sync_version)
         return "Rows: {0} ({1}), Total Execution Time: {2}. ({3:.2f} rows per second) ".format(self.total_row_count,
                                                                                                load_type,
                                                                                                self.total_execution_time,
@@ -89,6 +91,7 @@ def load_completed_successfully(self):
                 self.load_rows_per_second = self.row_count / self.load_execution_time.total_seconds()
 
 
+        # TODO: remove
         def load_skipped_due_to_zero_rows(self):
             self.status = "Skipped - Zero Rows"
             self.load_completed = datetime.now()

modules/data_load_tracking/DataLoadExecution.py

@@ -0,0 +1,38 @@
+import logging
+from modules.data_load_tracking.DataLoadExecution import DataLoadExecution, Base
+
+
+class DataLoadTrackerRepository(object):
+    def __init__(self, session_maker, logger=None):
+        self.logger = logger or logging.getLogger(__name__)
+        self.session_maker = session_maker
+
+    def create_tables(self, engine):
+        engine.execute("CREATE SCHEMA IF NOT EXISTS {0}".format("data_pipeline"))
+        Base.metadata.create_all(engine)
+
+    def get_last_sync_version(self, model_name):
+        session = self.session_maker()
+        result = session.query(DataLoadExecution).filter_by(model_name=model_name, status="Load Completed Successfully").order_by(DataLoadExecution.completed_on).first()
+
+        if result is None:
+            return 0
+        return result.next_sync_version
+
+
+    def save(self, data_load_tracker):
+
+        data_load_execution = DataLoadExecution(model_name=data_load_tracker.model_name,
+                                                correlation_id=data_load_tracker.correlation_id,
+                                                is_full_refresh=data_load_tracker.is_full_refresh,
+                                                this_sync_version=data_load_tracker.change_tracking_info.this_sync_version,
+                                                next_sync_version=data_load_tracker.change_tracking_info.next_sync_version,
+                                                execution_time_ms=int(data_load_tracker.total_execution_time.total_seconds() * 1000),
+                                                rows_processed=data_load_tracker.total_row_count,
+                                                status=data_load_tracker.status)
+
+
+
+        session = self.session_maker()
+        session.add(data_load_execution)
+        session.commit()

modules/DataLoadTracker.py …es/data_load_tracking/DataLoadTracker.pymodules/DataLoadTracker.py renamed to modules/data_load_tracking/DataLoadTracker.py modules/DataLoadTracker.py renamed to modules/data_load_tracking/DataLoadTracker.py

@@ -34,7 +34,6 @@ def prefix_column(column_name, full_refresh, primary_key_column_name):
     def build_select_statement(self, table_configuration, columns, batch_configuration, previous_batch_key, full_refresh, change_tracking_info):
         column_array = list(map(lambda cfg: self.prefix_column(cfg['source_name'], full_refresh, table_configuration['primary_key']), columns))
         column_names = ", ".join(column_array)
-        column_names = "{0}, {1} as data_pipeline_next_change_minimum_version".format(column_names, change_tracking_info.next_sync_version)
         if full_refresh:
             return "SELECT TOP ({0}) {1} FROM {2}.{3} t WHERE t.{4} > {5} ORDER BY t.{4}".format(batch_configuration['size'],
                                                                                            column_names,