""" Accounting Statement Export workflow ================================== Extract data for accounting statement export into schematized files on S3 location: s3://prod-statement-detail-exports/schematized_files. There are 4 sets of data for each month. They are: 1. Quarterly Labels 2. Quarterly Subaccounts 3. Monthly Labels 4. Monthly Subaccounts Example data set for accounting period 199: 1. /199,200,201_label_all_all_quarter 2. /199,200,201_subaccount_all_all_quarter 3. /199_label_all_all_month 4. /199_subaccount_all_all_month """ from garcon import runner from garcon import task from garcon.param import StaticParam from garcon_contrib.aws import emr from garcon_contrib.aws import garcon_s3 from processing_accounting.flows.accounting_statement_export import setting from processing_accounting.flows.accounting_statement_export import tasks from processing_accounting.flows.base import FlowBase from processing_accounting.util import emr as emr_util class Flow(FlowBase): required_params = ['period_ids', 'user_type'] available_params = [ 'period_ids', 'user_id', 'user_type', 'transaction_types', 'payment_interval', 'skip_unload'] def __init__(self): super().__init__('accounting_statement_export', '2.0') self.timeout = 21600 # override default to 6 hours (60 * 60 * 6) self.env = setting.env if setting.env_processing_accounting_sns_topic: self.sns_topic = setting.env_processing_accounting_sns_topic def workflow_id(self, initial_context=None): return '{flow_name}-{params}'.format( flow_name=self.name, params='_'.join(sorted([v for k, v in initial_context.items()]))) def decider(self, schedule, context): """Activity decider. Args: schedule (callable): the scheduler method. """ bootstrap = schedule('bootstrap', self.bootstrap) if setting.UPDATE_STATUS_ONLY in context: schedule( 'set_status', self.set_status, requires=[bootstrap]) else: if setting.SKIP_UNLOAD in context: launch_emr_cluster = schedule( 'launch_emr_cluster', self.launch_emr_cluster, requires=[bootstrap]) else: get_unload_query = schedule( 'get_unload_query', self.get_unload_query, requires=[bootstrap]) if get_unload_query.result.get('get_unload_query.stop'): return unload_data = schedule( 'unload_data_from_snowflake', self.unload_data_from_snowflake, requires=[get_unload_query]) launch_emr_cluster = schedule( 'launch_emr_cluster', self.launch_emr_cluster, requires=[unload_data]) hive_conversion_step = schedule( 'convert_tsv_to_avro', self.conversion_hive_step, requires=[launch_emr_cluster]) copy_from_hdfs_to_s3 = schedule( 'copy_from_hdfs_to_s3', self.copy_from_hdfs_to_s3, requires=[hive_conversion_step]) schedule( 'wait_for_emr_to_finish', self.wait_for_emr_to_finish, requires=[copy_from_hdfs_to_s3]) @property def bootstrap(self): """Bootstrap initial configuration. """ return self.create( name='bootstrap_config', retry=10, tasks=runner.Sync( tasks.bootstrap.fill( namespace='bootstrap', period_ids='period_ids', user_type='user_type', user_id='user_id', transaction_types='transaction_types', payment_interval='payment_interval'))) @property def get_unload_query(self): return self.create( name='get_unload_query', retry=10, tasks=runner.Sync( garcon_s3.remove_files_from_path.fill( path='bootstrap.raw_files_path'), garcon_s3.remove_files_from_path.fill( path='bootstrap.schematized_file_path'), tasks.get_unload_query.fill( namespace='get_unload_query', period_ids='period_ids', user_type='user_type', user_id='user_id', transaction_types='transaction_types', payment_interval='bootstrap.payment_interval'))) @property def unload_data_from_snowflake(self): return self.create( name='unload_data_from_snowflake', tasks=runner.Sync( tasks.snowflake_unload.fill( query='get_unload_query.unload_query', s3_prefix='bootstrap.raw_files_path', aws_access_key=StaticParam( setting.env_aws_access_key_id), aws_access_secret=StaticParam( setting.env_aws_secret_access_key)))) @property def launch_emr_cluster(self): return self.create( name='launch_emr_cluster', tasks=runner.Sync( tasks.generate_hql.fill( namespace='generate_hql', hash_key='bootstrap.hive_table_key'), emr.get_emr_cluster_by_tag.fill( tag_name=StaticParam( setting.config.get('emr').get('emr.tag_name'))), emr_util.launch_emr_cluster.fill( emr_cluster_id='emr.cluster_id', path_to_logs='bootstrap.emr.logs', job_name=StaticParam('Accounting Statement Export'), tag_name=StaticParam( setting.config.get('emr').get('emr.tag_name')), ec2_instance_name=StaticParam( setting.config.get('emr').get( 'emr.ec2_instance_name')), keep_alive=StaticParam(False)), emr_util.add_hive_settings_step.fill( emr_cluster_id='emr.cluster_id', step_name=StaticParam('Modify Hive Settings'), step_args=StaticParam([ 's3://elasticmapreduce/libs/hive/hive-script', '--base-path', 's3://elasticmapreduce/libs/hive/', '--run-hive-script', '--args', "-e", "set hive.stats.autogather=false;"])))) @property def conversion_hive_step(self): return self.create( name='conversion_hive_step', tasks=runner.Sync( emr_util.add_hive_job_step.fill( namespace='add_hive_job_step', emr_cluster_id='emr.cluster_id', source_s3_path='bootstrap.conversion_hive_step_input', destination_s3_path='bootstrap.hdfs_staging_path', step_name=StaticParam('Generate Avro on HDFS'), hql_s3_path='generate_hql.hql_path'))) @property def copy_from_hdfs_to_s3(self): return self.create( name='copy_from_hdfs_to_s3', tasks=runner.Sync( emr_util.add_s3_distcp_step.fill( namespace='copy_from_hdfs_to_s3', emr_cluster_id='emr.cluster_id', source_path='bootstrap.hdfs_staging_path', destination_path='bootstrap.schematized_file_path'))) @property def wait_for_emr_to_finish(self): return self.create( name='wait_for_emr_to_finish', tasks=runner.Sync( task.decorate(60*60*3)( emr.wait_emr_step_completed.fill( namespace='wait emr step completed', emr_cluster_id='emr.cluster_id', emr_step_id='copy_from_hdfs_to_s3.emr.step_id')))) @property def set_status(self): return self.create( name='set status', tasks=runner.Sync( tasks.update_status.fill( namespace='update_status', period_ids='period_ids', user_type='user_type', payment_interval='bootstrap.payment_interval', user_id='bootstrap.user_id', status=StaticParam('GENERATED'))))