""" Deezer Ingestion Workflow. Ingest data from Deezer v2 Feed. """ from garcon.param import StaticParam from garcon_contrib.dynamo_feed_status import garcon_feed_status from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.flows import base from feed_ingestion.flows.deezer import config from feed_ingestion.flows.deezer import tasks from feed_ingestion.flows.deezer.stage_loader import DeezerSL from feed_ingestion.tasks import deezer_tasks, load_fact_tables_tasks_sf from feed_ingestion.tasks import overall_status_tasks from feed_ingestion.util.jenkins.tasks import build_jenkins_dbt class Flow(base.FlowLicensor, base.FlowConfigMixin, base.FlowLoadRawMixinSF, base.FlowLoadFactMixinSF): """Class representing the workflow.""" def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(feed_name=config.feed_name, version='3.0') self.timeout = 60 * 60 * 5 # set "Execution Start To Close Timeout" self.GRAB_DROP_FILES = { 'sme_v2': self.grab_drop_files_sme, 'sme_v3': self.grab_drop_files_sme, 'theorchard_v1': self.fetch_from_drop_location, 'theorchard_v3': self.grab_drop_files_sme, 'altafonte_v4': self.grab_drop_files_altafonte } def decider(self, schedule, context): """Activity decider. Args: schedule (callable): The scheduler method. """ bootstrap = schedule('bootstrap', self.bootstrap) # Stop flow if feed already ingested if bootstrap.result.get('bootstrap.stop'): return is_fraud_backfill = bootstrap.result.get( 'bootstrap.fraud_report_backfill') if is_fraud_backfill: grab_fraud = schedule( 'grab_fraud_report_backfill', self.grab_fraud_report_backfill, requires=[bootstrap]) if grab_fraud.result.get( 'grab_fraud_report_backfill.stop'): return schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[grab_fraud]) return version = bootstrap.result.get('bootstrap.spec_version') if version is None: if bootstrap.result.get('bootstrap.licensor') == 'theorchard': version = 1 else: version = 2 licensor_version = '_v'.join([ bootstrap.result.get('bootstrap.licensor'), str(version) ]) grab_drop_files = schedule( 'grab_drop_files', self.GRAB_DROP_FILES[licensor_version], requires=[bootstrap]) if grab_drop_files.result.get('grab_drop_files.stop'): return # Status set to DOWNLOADED if files are present, else NOT_AVAILABLE set_status_to_downloaded = schedule( 'set_status_to_downloaded', self.set_status_to_downloaded, requires=[grab_drop_files]) if bootstrap.result.get('bootstrap.use_s3'): next_steps_requires = [set_status_to_downloaded] else: unzip_and_clean = schedule( 'unzip_and_clean', self.unzip_and_clean, requires=[set_status_to_downloaded]) next_steps_requires = [unzip_and_clean] load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=next_steps_requires) # Status updated to POPULATED_RAW_TABLE # if staging_raw_only flag passed in initial context, skip fact load if 'staging_raw_only' in context: return if bootstrap.result.get('bootstrap.licensor') == 'altafonte': return if bootstrap.result.get('bootstrap.spec_version') >= 2: update_dim_tables = schedule( 'update_dim_tables', self.update_dim_tables, requires=[load_staging_raw_table]) last_activity = update_dim_tables else: last_activity = load_staging_raw_table # Load from staging_raw to fact_analytics & fact_analytics_error load_staging_fact_table = schedule( 'load_staging_fact_table', self.load_staging_fact_table, requires=[last_activity]) load_fact_tables = schedule( 'load_fact_tables', self.load_fact_tables, requires=[load_staging_fact_table]) # Status updated to INGESTED schedule( 'build_jenkins_dbt', self.build_jenkins_dbt, requires=[load_fact_tables]) @property def bootstrap(self): """Bootstrap initial configuration.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='context_date', licensor='licensor', reload='reload', use_s3='use_s3', date_format='date_format', fraud_report_backfill='fraud_report_backfill'))) @property def grab_fraud_report_backfill(self): """Copy fraud report from separate fraudulent_reports folder.""" return self.create( name='grab_fraud_report_backfill', tasks=base.AsyncRunner( tasks.grab_fraud_report_backfill.fill( namespace='grab_fraud_report_backfill', feed_name='bootstrap.feed_name', date='bootstrap.date', source_bucket_name='bootstrap.s3_drop_bucket', destination_full_path=( 'bootstrap.s3_temp_staging_raw_bucket'), licensor='bootstrap.licensor', ))) @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', tasks=base.AsyncRunner( tasks.grab_drop_files_sme.fill( namespace='grab_drop_files', feed_name='bootstrap.feed_name', date='bootstrap.date', drop_file_name='bootstrap.drop_file_name', source_bucket_name='bootstrap.s3_drop_bucket', source_path='bootstrap.source_path', destination_full_path='bootstrap.s3_archive_bucket', ))) @property def grab_drop_files_altafonte(self): """Grab files and upload to archive bucket on S3 for sme.""" return self.create( name='grab_drop_files_altafonte', tasks=base.AsyncRunner( tasks.grab_drop_files_altafonte.fill( namespace='grab_drop_files', feed_name='bootstrap.feed_name', date='bootstrap.date', drop_file_name='bootstrap.drop_file_name', source_bucket_name='bootstrap.s3_drop_bucket', source_path='bootstrap.source_path', destination_full_path='bootstrap.s3_archive_bucket', source_files_dict='bootstrap.source_files_dict', ))) @property def fetch_from_drop_location(self): """Move file from Zephir to drop bucket.""" return self.create( name='fetch_from_drop_location', tasks=base.SyncRunner( tasks.fetch_from_drop_location.fill( namespace='grab_drop_files', feed_name='bootstrap.feed_name', date='bootstrap.date', source_path='bootstrap.source_path', target_s3_path='bootstrap.s3_archive_bucket', drop_file_name='bootstrap.drop_file_name'))) @property def unzip_and_clean(self): """Unzip drop file and clean then place to temp path.""" return self.create( name='unzip_and_clean', tasks=base.SyncRunner( deezer_tasks.unzip_and_clean.fill( namespace='unzip_and_clean', feed_name='bootstrap.feed_name', date='bootstrap.date', source_s3_path='bootstrap.s3_archive_bucket', target_s3_path='bootstrap.s3_temp_staging_raw_bucket', drop_file_name='bootstrap.drop_file_name', licensor='bootstrap.licensor', source_files_dict='bootstrap.source_files_dict', ))) @property def load_fact_tables(self): """Activity to load fact_analytics & fact_analytics_error. Deletes any existing data for this feed and date in fact_analytics & fact_analytics_error. After successfully loading the data, sets the feed's status for that download_date to INGESTED. """ activity_name = 'load_fact_tables' return self.create( name=activity_name, tasks=base.SyncRunner( load_fact_tables_tasks_sf.load_fact_data.fill( namespace='load_fact_data', feed_name='{}.feed_name'.format(self.bootstrap_namespace), secrets_path='{}.secrets_path'.format( self.bootstrap_namespace), date='{}.date'.format(self.bootstrap_namespace), date_as_in_uuid='{}.date_as_in_uuid'.format( self.bootstrap_namespace), sfdb_params='sfdb_params', kwargs='{}.common_kwargs'.format( self.bootstrap_namespace)), overall_status_tasks.set_overall_status.fill( feed_name='{}.feed_name'.format(self.bootstrap_namespace), date='{}.date'.format(self.bootstrap_namespace), set_status_once=StaticParam(True), status=StaticParam(garcon_feed_status.STATUS_INGESTED)))) load_staging_raw_table = DeezerSL.load_activity( feed_name='bootstrap.feed_name', secrets_path=config.secrets_path, sql_loader=SQLLoader(__file__), requirements=dict( date='bootstrap.date', feed_name='bootstrap.feed_name', source_files_dict='bootstrap.source_files_dict', s3_dir_path='bootstrap.s3_temp_staging_raw_bucket', staging_raw_table_name='bootstrap.staging_raw_table', licensor='bootstrap.licensor', date_format='bootstrap.date_format')) @property def update_dim_tables(self): """Update dimension tables (dim_user and dim_playlist).""" 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='sfdb_params', kwargs='bootstrap.dimension_tables'), tasks.sns_publish_message.fill( topic=StaticParam( config.dimension_tables['sns_topic']), feed_name='bootstrap.feed_name', date='bootstrap.date', subject='update_dim_tables.sns_report_subject', message='update_dim_tables.sns_report_message'))) @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 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-deezer-exec/ # noqa config='bootstrap.jenkins_config')))