actions_taken_to_redshift.py

actions_taken_to_redshift.py

import luigi
from buffer_redshift_etl import *

class ExtractActionsTakenTask(BufferRedshiftETLExtractTask):
    def etl_name(self):
        return 'actions-taken'

class TransformActionsTakenTask(BufferRedshiftETLTransformTask):
    def etl_name(self):
        return 'actions-taken'

    def extract_class(self):
        return ExtractActionsTakenTask

class CopyActionsTakenToRedshiftTask(BufferCopyToRedshiftTask):
    def etl_name(self):
        return 'actions-taken'

    def table_name(self):
        return 'actions_taken'

    @property
    def columns(self):
        return [
          ('user_id', 'text'),
          ('visitor_id', 'text'),
          ('client_id', 'text'),
          ('last_modified', 'timestamp'),
          ('user_joined_at', 'timestamp'),
          ('date', 'timestamp'),
          ('scope1', 'text'),
          ('scope2', 'text'),
          ('scope3', 'text'),
          ('scope4', 'text'),
          ('scope5', 'text'),
          ('scope6', 'text'),
          ('scope7', 'text'),
          ('scope8', 'text'),
          ('scope9', 'text'),
          ('scope10', 'text'),
          ('extra_data', 'varchar (max)'),
          ('PRIMARY KEY', '(user_id, date)')
        ]

    def transform_class(self):
        return TransformActionsTakenTask

class ShutdownClusters(BufferShutdownClusters):
    def copy_to_redshift_class(self):
        return CopyActionsTakenToRedshiftTask

class ActionsTakenETLTask(BufferETLTask):
    def copy_to_redshift_class(self):
        return CopyActionsTakenToRedshiftTask

    def shutdown_class(self):
        return ShutdownClusters

if __name__ == "__main__":
    luigi.run(main_task_cls=ActionsTakenETLTask)