Inital commit from Oscars-Misc

Author: dames

.gitignore

@@ -1,3 +1,5 @@
+.idea/
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

BatchDataLoader.py

@@ -0,0 +1,100 @@
+import logging
+import pandas
+from io import StringIO
+import importlib
+
+
+
+
+
+class BatchDataLoader(object):
+    def __init__(self, source_table_configuration, columns, batch_configuration, logger=None):
+        self.logger = logger or logging.getLogger(__name__)
+        self.source_table_configuration = source_table_configuration
+        self.columns = columns
+        self.batch_configuration = batch_configuration
+
+    def build_select_statement(self, previous_key=0):
+
+        column_array = list(map(lambda cfg: cfg['source_name'], self.columns))
+        column_names = ", ".join(column_array)
+
+
+        return "SELECT TOP ({0}) {1} FROM {2}.{3} WHERE {4} > {5} ORDER BY {4}".format(self.batch_configuration['size'],
+                                                            column_names,
+                                                            self.source_table_configuration['schema'],
+                                                            self.source_table_configuration['name'],
+                                                            self.batch_configuration['source_unique_column'],
+                                                            previous_key
+                                                            )
+
+    # Imports rows, returns True if >0 rows were found
+    def import_batch(self, source_engine, target_engine, target_table_configuration, batch_tracker, previous_key):
+        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")
+        data_frame = pandas.read_sql_query(sql, source_engine)
+        self.logger.info("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
+
+        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))
+
+        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))
+        data = StringIO()
+        data_frame.to_csv(data, header=False, index=False)
+        data.seek(0)
+        raw = target_engine.raw_connection()
+        curs = raw.cursor()
+
+        column_array = list(map(lambda cfg: cfg['destination']['name'], self.columns))
+
+        curs.copy_from(data, destination_table, sep=',', columns=column_array)
+        self.logger.info("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)
+                #transformer = self.create_column_transformer_type(column['column_transformer'])
+               #// df['a'] = df['a'].map(lambda a: a / 2.)
+
+                #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

DataLoadManager.py

@@ -0,0 +1,64 @@
+import logging
+from BatchDataLoader import BatchDataLoader
+from DestinationTableManager import DestinationTableManager
+from DataLoadTracker import DataLoadTracker
+from SourceTableManager import SourceTableManager
+import json
+
+
+class DataLoadManager(object):
+    def __init__(self, configuration_path, logger=None):
+        self.logger = logger or logging.getLogger(__name__)
+        self.configuration_path = configuration_path
+
+    def start_import(self, source_engine, target_engine, configuration_name, full_load):
+
+        with open("{0}{1}.json".format(self.configuration_path, configuration_name)) as json_data:
+            pipeline_configuration = json.load(json_data)
+
+        data_load_tracker = DataLoadTracker(configuration_name, json_data, full_load)
+
+        self.logger.debug("Execute Starting")
+        destination_table_manager = DestinationTableManager()
+
+        columns = self.remove_invalid_columns(pipeline_configuration['source_table'], pipeline_configuration['columns'], source_engine)
+
+        if full_load:
+            self.logger.info("Full-load is set. Recreating the staging table.")
+            destination_table_manager.create_table(pipeline_configuration['stage_table'],
+                                                   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'])
+
+        previous_unique_column_value = 0
+        while previous_unique_column_value > -1:
+            previous_unique_column_value = batch_importer.import_batch(source_engine, target_engine, pipeline_configuration['stage_table'], data_load_tracker.start_batch(), previous_unique_column_value)
+
+
+        self.logger.info("ImportBatch Completed")
+
+        #if full_load:
+            #return
+            # Rename the stage table to the load table.
+            # log.information("Full-load is set. Renaming the stage table to the load table.")
+            # rename_table(pipeline_configuration['stage_source_data'], pipeline_configuration['load_source_data'])
+        #else:
+            #return
+            # upsert_data_from_stage_to_load_tables(pipeline_configuration['stage_source_data'], pipeline_configuration['load_source_data'])
+
+        data_load_tracker.completed_successfully()
+        self.logger.info(data_load_tracker.get_statistics())
+
+    def remove_invalid_columns(self, source_table_configuration, column_configration, source_engine):
+        source_table_manager = SourceTableManager()
+        existing_columns = source_table_manager.get_columns(source_table_configuration, source_engine)
+        return list(filter(lambda column: self.column_exists(column['source_name'], existing_columns), column_configration))
+
+    def column_exists(self, column_name, existing_columns):
+        if column_name in existing_columns:
+            return True
+        self.logger.warning("Column {0} does not exist in source. It will be ignored for now, however may cause downstream issues.".format(column_name))
+        return False

DataLoadTracker.py

@@ -0,0 +1,62 @@
+from datetime import datetime
+
+
+class DataLoadTracker:
+    started = datetime.now()
+    completed = None
+    status = "Not Started"
+    total_row_count = 0
+    batches = []
+    configuration_name = None
+    configuration = None
+    is_full_load = False
+    total_execution_time = None
+    total_row_count = 0
+    rows_per_second = 0
+
+    def __init__(self, configuration_name, configuration, is_full_load):
+        self.configuration_name = configuration_name
+        self.configuration = configuration
+        self.is_full_load = is_full_load
+
+    def start_batch(self):
+        batch = self.Batch()
+        self.batches.append(batch)
+        return batch
+
+    def completed_successfully(self):
+        self.completed = datetime.now()
+        self.total_execution_time = self.completed - self.started
+
+        for batch in self.batches:
+            self.total_row_count = self.total_row_count + batch.row_count
+
+        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)
+
+    class Batch:
+        row_count = 0
+        extract_started = datetime.now()
+        extract_completed_on = None
+        load_completed_on = None
+        status = "Not Started"
+
+        def __init__(self):
+            pass
+
+        def extract_completed_successfully(self, row_count):
+            self.status = "Extract Completed Successfully"
+            self.row_count = row_count
+            self.extract_completed_on = datetime.now()
+
+        def load_completed_successfully(self):
+            self.status = "Load Completed Successfully"
+            self.load_completed_on = datetime.now()
+
+        def load_skipped_due_to_zero_rows(self):
+            self.status = "Skipped - Zero Rows"
+            self.load_completed_on = datetime.now()

DestinationTableManager.py

@@ -0,0 +1,52 @@
+import logging
+from sqlalchemy import MetaData, DateTime
+from sqlalchemy.schema import Column, Table
+import importlib
+from sqlalchemy.sql import func
+
+
+class DestinationTableManager(object):
+    def __init__(self, logger=None):
+        self.logger = logger or logging.getLogger(__name__)
+
+    def create_table(self, table_configuration, columns_configuration, target_engine, drop_first):
+        metadata = MetaData()
+
+        table = Table(table_configuration['name'], metadata, schema=table_configuration['schema'])
+
+        for column_configuration in columns_configuration:
+            table.append_column(self.create_column(column_configuration['destination']))
+
+        table.append_column(
+            Column("data_pipeline_timestamp", DateTime(timezone=True), server_default=func.now()))
+
+        if drop_first:
+            self.logger.info(
+                "Dropping table {0}.{1}".format(table_configuration['name'], table_configuration['schema']))
+            table.drop(target_engine, checkfirst=True)
+            self.logger.debug(
+                "Dropped table {0}.{1}".format(table_configuration['name'], table_configuration['schema']))
+
+        self.logger.info("Creating table {0}.{1}".format(table_configuration['name'], table_configuration['schema']))
+        table.create(target_engine, checkfirst=False)
+        return
+
+    def create_column_type(self, type_name):
+        parts = type_name.split(".")
+        module = importlib.import_module(parts[0])
+        class_ = getattr(module, parts[1])
+        instance = class_()
+        return instance
+
+    def create_column(self, configuration):
+        return Column(configuration['name'], self.create_column_type(configuration['type']),
+                      primary_key=configuration.get("primary_key", False),
+                      nullable=configuration['nullable']);
+
+    def rename_table(self, source_table_configuration, target_table_configuration):
+        print('TODO - create a rename-table method. Eg: ALTER TABLE table_name RENAME TO new_table_name;')
+        return;
+
+    def upsert_data_from_stage_to_load_tables(self, source_table_configuration, target_table_configuration):
+        print('TODO - create a method to upsert the data;')
+        return;

SourceTableManager.py

@@ -0,0 +1,18 @@
+import logging
+from sqlalchemy import MetaData
+from sqlalchemy.schema import Table
+
+
+
+class SourceTableManager(object):
+    def __init__(self, logger=None):
+        self.logger = logger or logging.getLogger(__name__)
+
+    def get_columns(self, table_configuration, source_engine):
+        metadata = MetaData()
+        self.logger.debug("Reading definition for source table {0}.{1}".format(table_configuration['schema'], table_configuration['name']))
+        table = Table(table_configuration['name'], metadata, schema=table_configuration['schema'], autoload=True, autoload_with=source_engine)
+        return list(map(lambda column:column.name, table.columns))
+
+
+

column_transformers/ToUpper.py

@@ -0,0 +1,4 @@
+class TextCleanUp:
+    def execute(text_in):
+        return text_in.upper()
+