""" Pandora Analytics Ingestion Workflow. Ingests Pandora data and loads into fact analytics. """ import datetime from garcon.param import StaticParam from garcon_contrib.aws import garcon_sns from garcon_contrib.aws.utils import garcon_s3 from garcon_contrib.dynamo_feed_status import garcon_feed_status from feed_ingestion.flows import base from feed_ingestion.flows.base import FlowBase from feed_ingestion.flows.base import FlowConfigMixin from feed_ingestion.flows.base import FlowLoadFactMixinSF from feed_ingestion.flows.pandora import config from feed_ingestion.flows.pandora import tasks from feed_ingestion.tasks import feed_status_tasks, validate_raw_data_tasks_sf from feed_ingestion.tasks import load_fact_tables_tasks_sf from feed_ingestion.tasks import load_raw_table_tasks_sf from feed_ingestion.tasks import overall_status_tasks from feed_ingestion.tasks import s3_tasks from feed_ingestion.util.jenkins.tasks import build_jenkins_dbt class Flow(FlowBase, FlowConfigMixin, FlowLoadFactMixinSF): """Pandora Flow class. This flow downloads the Pandora data and loads it into staging_raw_pandora and fact_analytics/fact_analytics_error tables. """ def __init__(self): """Initialize a Pandora flow.""" super(Flow, self).__init__(config.feed_name, config.feed_version) self.GRAB_DROP_FILES = { 'sme': self.grab_drop_files_sme, 'theorchard': self.grab_drop_files } def decider(self, schedule): """Activity decider. Args: schedule (callable): The scheduler method. """ # get execution parameters bootstrap = schedule( 'bootstrap', self.bootstrap) # stop flow if data already ingested if bootstrap.result.get('bootstrap.stop') is True: return grab_drop_files = schedule( 'grab_drop_files', self.GRAB_DROP_FILES[bootstrap.result.get('bootstrap.licensor')], requires=[bootstrap]) # update file statuses update_feed_s3_file_status = schedule( 'update_feed_s3_file_status', self.update_feed_s3_file_status, requires=[grab_drop_files]) # stop flow if any file is unavailable file_status = update_feed_s3_file_status.result.get( 'update_feed_s3_file_status.file_status') if file_status == garcon_feed_status.STATUS_NOT_AVAILABLE: return # set overall feed status to DOWNLOADED set_status_to_downloaded = schedule( 'set_feed_status_downloaded', self.set_status_to_downloaded, requires=[update_feed_s3_file_status]) # clean out old data clean_staging_raw = schedule( 'clean_staging_raw', self.clean_staging_raw, requires=[set_status_to_downloaded]) # populate temporary staging tables populate_temp_staging_tables = schedule( 'populate_temp_staging_tables', self.populate_temp_staging_tables, requires=[clean_staging_raw]) # load staging raw table populate_staging_raw = schedule( 'populate_staging_raw', self.populate_staging_raw, requires=[populate_temp_staging_tables]) mark_staging_raw_table_tasks_complete = schedule( 'mark_staging_raw_table_tasks_complete', self.mark_staging_raw_table_tasks_complete, requires=[populate_staging_raw]) # set overall feed status to POPULATED_RAW_TABLE set_status_to_populated_raw_table = schedule( 'set_status_to_populated_raw_table', self.set_status_to_populated_raw_table, requires=[mark_staging_raw_table_tasks_complete]) # drop temporary staging tables drop_temp_staging_tables = schedule( 'drop_temp_staging_tables', self.drop_temp_staging_tables, requires=[set_status_to_populated_raw_table]) update_dim_tables = schedule( 'update_dim_tables', self.update_dim_tables, requires=[drop_temp_staging_tables]) if update_dim_tables.result.get( 'update_dim_tables.sns_report_subject', False): schedule( 'send_dimension_tables_update_report_sns_notification', self.send_dimension_tables_update_report_sns_notification, requires=[update_dim_tables]) # load from staging_raw to staging_fact table load_staging_fact_table = schedule( 'load_staging_fact_table', self.load_staging_fact_table, requires=[update_dim_tables]) # load fact data into Snowflake load_fact_tables = schedule( 'load_fact_tables', self.load_fact_tables, requires=[load_staging_fact_table]) schedule( 'build_jenkins_dbt', self.build_jenkins_dbt, requires=[load_fact_tables]) def workflow_id(self, initial_context): """Generate workflow id. Args: initial_context (dict): The initial context for the flow. Returns: str: A unique identifier for a workflow being executed. """ licensor = initial_context['licensor'] flow_name = '_'.join([self.name, licensor]) if 'context_date' not in initial_context: date = datetime.datetime.today().strftime('%Y-%m-%d') else: date = initial_context['context_date'] return '{flow_name}-{date}'.format( flow_name=flow_name, date=date) def contextified_feed_name(self, context): """Get feed_name in context. Args: context (dict): The context of the flow. Returns: str: Contextified feed name. """ assert 'licensor' in context, 'There is no licensor in context' assert context['licensor'] in config.licensors, \ 'There is no such licensor in config file' return '_'.join([self.feed_name, context['licensor']]) @property def bootstrap(self): """Bootstrap initial configuration.""" seoccm = 'snowflake_error_on_column_count_mismatch' return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='context_date', reload='reload', licensor='licensor', snowflake_error_limit='snowflake_error_limit', snowflake_error_on_column_count_mismatch=seoccm ))) @property def grab_drop_files(self): """Grab files and upload to archive bucket on S3 for theorchard.""" return self.create( name='grab_drop_files', generators=[self.drop_files_generator], tasks=base.AsyncRunner( s3_tasks.copy_file.fill( namespace='copy_file', source_bucket_name=StaticParam(config.drop_bucket), source_key_name='source_key_name', destination_bucket_name=StaticParam(config.data_bucket), destination_key_name='destination_key_name', replace='bootstrap.replace_archive_files'), max_workers=4)) @property def grab_drop_files_sme(self): """Grab files and upload to archive bucket on S3 for sme.""" return self.create( name='grab_drop_files_sme', generators=[self.drop_files_generator], tasks=base.AsyncRunner( s3_tasks.copy_file_from_sme_s3_to_theocrhard.fill( namespace='grab_drop_files_sme', secrets_path=StaticParam(config.secrets_path), source_bucket_name=StaticParam(config.sme_drop_bucket), source_key_name='source_key_name', destination_bucket_name=StaticParam(config.data_bucket), destination_key_name='destination_key_name', replace='bootstrap.replace_archive_files'), max_workers=4)) @property def update_feed_s3_file_status(self): """Update status of downloaded files.""" return self.create( name='update_feed_s3_file_status', tasks=base.SyncRunner( feed_status_tasks.update_feed_s3_file_status.fill( namespace='update_feed_s3_file_status', s3_path='bootstrap.archive_bucket', file_names='bootstrap.expected_files', feed_name='bootstrap.feed_name', date='bootstrap.date'))) @property def set_status_to_downloaded(self): """Set overall feed status to DOWNLOADED.""" return self.create( name='set_status_to_downloaded', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_downloaded', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam(garcon_feed_status.STATUS_DOWNLOADED)))) @property def clean_staging_raw(self): """Delete any existing data from staging raw table.""" return self.create( name='clean_staging_raw', tasks=base.SyncRunner( tasks.clean_staging_raw_table.fill( namespace='clean_staging_raw_table', date='bootstrap.date', feed_name='bootstrap.feed_name'))) @property def populate_temp_staging_tables(self): """Populate the temporary staging tables with the raw files.""" seoccm = 'bootstrap.snowflake_error_on_column_count_mismatch' return self.create( name='populate_temp_staging_tables', generators=[self.temp_staging_tables_generator], tasks=base.SyncRunner( tasks.create_temp_staging_raw_table.fill( namespace='create_temp_staging_raw_table', date='bootstrap.date', temp_table_name='temp_table_name', feed_name='bootstrap.feed_name'), validate_raw_data_tasks_sf.load_temp_staging_raw_table.fill( namespace='load_temp_staging_raw_table', date='bootstrap.date', sfdb_params='bootstrap.sfdb_params', feed_name='bootstrap.feed_name', secrets_path=StaticParam(config.secrets_path), kwargs='kwargs', key_dir='temp_table_s3_full_path', temp_staging_raw_table='temp_table_name', error_limit='bootstrap.snowflake_error_limit', snowflake_error_on_column_count_mismatch=seoccm ))) @property def populate_staging_raw(self): """Load staging raw table from temp tables.""" return self.create( name='populate_staging_raw', generators=[self.load_staging_raw_table_generator], tasks=base.SyncRunner( tasks.load_staging_raw_table.fill( namespace='load_staging_raw_table', date='bootstrap.date', processed_datetime='bootstrap.processed_datetime', filename='filename', temp_streams_table='temp_streams_table', temp_metadata_table='temp_metadata_table', staging_raw_table=StaticParam( config.snowflake_table_names['staging_raw']), feed_name='bootstrap.feed_name'))) @property def mark_staging_raw_table_tasks_complete(self): """Set the complete staging_raw_table_tasks to feed.""" return self.create( name='mark_staging_raw_table_tasks_complete', tasks=base.SyncRunner( (load_raw_table_tasks_sf.mark_staging_raw_table_tasks_complete. fill( namespace='mark_staging_raw_table_tasks_complete', date='bootstrap.date', feed_name='bootstrap.feed_name')))) @property def set_status_to_populated_raw_table(self): """Set overall feed status to POPULATED_RAW_TABLE.""" return self.create( name='set_status_to_populated_raw_table', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_populated_raw_table', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_POPULATED_RAW_TABLE)))) @property def drop_temp_staging_tables(self): """Drop the temporary staging tables.""" return self.create( name='drop_temp_staging_tables', generators=[self.temp_staging_tables_generator], tasks=base.SyncRunner( tasks.drop_temp_table.fill( namespace='drop_temp_table', temp_table_name='temp_table_name'))) @property def update_dim_tables(self): """Update dimension tables.""" return self.create( name='update_dim_tables', schedule_to_start=48000, tasks=base.SyncRunner( load_fact_tables_tasks_sf.update_dim_tables.fill( namespace='update_dim_tables', feed_name='bootstrap.feed_name', secrets_path=StaticParam(config.secrets_path), date='bootstrap.date', sfdb_params='bootstrap.sfdb_params', kwargs=StaticParam(config.dimension_tables)))) @property def send_dimension_tables_update_report_sns_notification(self): """Send SNS message with dimension tables update report.""" return self.create( name='send_dimension_tables_update_report_sns_notification', tasks=base.SyncRunner( garcon_sns.sns_publish_message.fill( topic=StaticParam( config.dimension_tables['sns_topic']), message='update_dim_tables.sns_report_message', subject='update_dim_tables.sns_report_subject'))) @property def build_jenkins_dbt(self): """Build Jenkins DBT job.""" return self.create( name='build_jenkins_dbt', schedule_to_start=48000, tasks=base.SyncRunner( build_jenkins_dbt.fill( namespace='build_jenkins_dbt', date='bootstrap.date', feed_name='bootstrap.feed_name', licensor='bootstrap.licensor', build_dbt='build_dbt', # coming from context of the https://scheduler.theorchard.io/job/swf-pandora-exec/ # noqa config='bootstrap.jenkins_config'))) def drop_files_generator(self, context): """Generate raw files to archive. Used by the grab_drop_files activity. Args: context (dict): The current context. Yields: dict: Dictionary of file to archive and archive destination. """ for file in context['bootstrap.expected_files']: drop_file = '{dir}{file}'.format( dir=context['bootstrap.drop_bucket'], file=file) archive_file = '{dir}{file}'.format( dir=context['bootstrap.archive_bucket'], file=file) file_status = garcon_feed_status.get_status( self.feed_name, context['bootstrap.date'], file) if file_status != garcon_feed_status.STATUS_DOWNLOADED: source_key_name = ( garcon_s3.extract_bucket_path(drop_file)[1]) destination_key_name = ( garcon_s3.extract_bucket_path(archive_file)[1]) yield dict( source_key_name=source_key_name, destination_key_name=destination_key_name) def temp_staging_tables_generator(self, context): """Generate parameters for temporary staging tables. Used by the populate_temp_staging_tables and populate_temp_staging_tables activity. Args: context (dict): The current context. Yields: dict: Dictionary of temporary staging table parameters. """ temp_tables = context['bootstrap.temp_staging_raw_tables'] for table in temp_tables.values(): table['kwargs'] = dict( error_on_column_count_mismatch=context[ 'bootstrap.' 'snowflake_error_on_column_count_mismatch'], ) yield table def load_staging_raw_table_generator(self, context): """Generate parameters for loading staging raw table. Args: context (dict): The current context. Yields: dict: Dictionary of parameters. """ temp_table_info = context['bootstrap.temp_staging_raw_tables'] temp_metadata_table = temp_table_info['metadata']['temp_table_name'] for country in config.countries: temp_streams_table = temp_table_info[country]['temp_table_name'] filename = temp_table_info[country][ 'temp_table_s3_full_path'].split('/')[-1] yield dict( filename=filename, temp_streams_table=temp_streams_table, temp_metadata_table=temp_metadata_table)