From 88b2a3835f8306dee842378a94d98dbbdfaaede6 Mon Sep 17 00:00:00 2001 From: Umang Mistry Date: Tue, 5 Mar 2024 15:42:32 -0600 Subject: [PATCH 1/2] Updated pipelines yaml with EC2 deployment + OIDC connections --- airflow/Qa_Dag.py | 13 +++++++++++++ bitbucket-pipelines.yml | 30 +++++++++++++++++++++++++++--- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/airflow/Qa_Dag.py b/airflow/Qa_Dag.py index bceed96..2896f1b 100644 --- a/airflow/Qa_Dag.py +++ b/airflow/Qa_Dag.py @@ -67,6 +67,19 @@ def getData(): print(f"Successful S3 put_object response. Status - {status}") else: raise AirflowFailException(f"Unsuccessful S3 put_object response. Status - {status}") + + oldData = df + newData = df + + with io.BytesIO() as output: + with pd.ExcelWriter(output, engine='xlsxwriter') as writer: + oldData.to_excel(writer, sheet_name="Old") + newData.to_excel(writer, sheet_name="new") + dat = oldData.compare(newData, align_axis=0, keep_shape=True) + dat.to_excel(writer, sheet_name="qc") + response = s3.put_object( + Bucket=bucket, Key=object_key+'QC_Report.xlsx' , Body=output.getvalue() + ) print("Dataframe is written to S3 successfully.") diff --git a/bitbucket-pipelines.yml b/bitbucket-pipelines.yml index 16e644c..e8132ab 100644 --- a/bitbucket-pipelines.yml +++ b/bitbucket-pipelines.yml @@ -21,10 +21,27 @@ definitions: - if [ "$BITBUCKET_BRANCH" == "UAT" ]; then export SNOWFLAKE_DATABASE=$UAT_SNOWFLAKE_DATABASE; export SNOWFLAKE_ROLE=$UAT_SNOWFLAKE_ROLE; fi - if [ "$BITBUCKET_BRANCH" == "PROD" ]; then export SNOWFLAKE_DATABASE=$PROD_SNOWFLAKE_DATABASE; export SNOWFLAKE_ROLE=$PROD_SNOWFLAKE_ROLE; fi - schemachange -a $SNOWFLAKE_ACCOUNT -u $SNOWFLAKE_USER -r $SNOWFLAKE_ROLE -w $SNOWFLAKE_WAREHOUSE -d $SNOWFLAKE_DATABASE -c $SNOWFLAKE_DATABASE.SCHEMACHANGE.CHANGE_HISTORY --create-change-history-table -v - trigger: automatic # uncomment after testing + trigger: automatic # trigger: manual caches: - pip + - step: &streamlit_deploy + name: "Deploy streamlit to EC2" + image: python:3.8 + script: + - python -m pip install --upgrade pip + - apt-get update && apt-get install -y jq git + - pip install awscli + - export AWS_ROLE_ARN=arn:aws:iam::$AWS_ACCOUNT_NO:role/$OIDC_ROLE + - export AWS_WEB_IDENTITY_TOKEN_FILE=$(pwd)/web-identity-token + - echo $BITBUCKET_STEP_OIDC_TOKEN > $(pwd)/web-identity-token + - export STS_OUTPUT=$(aws sts assume-role-with-web-identity --role-arn $AWS_ROLE_ARN --role-session-name BitbucketPipeline --web-identity-token "$BITBUCKET_STEP_OIDC_TOKEN" --duration-seconds 3600) + - export AWS_ACCESS_KEY_ID=$(echo $STS_OUTPUT | jq -r '.Credentials.AccessKeyId') + - export AWS_SECRET_ACCESS_KEY=$(echo $STS_OUTPUT | jq -r '.Credentials.SecretAccessKey') + - export AWS_SESSION_TOKEN=$(echo $STS_OUTPUT | jq -r '.Credentials.SessionToken') + - aws sts get-caller-identity + - aws ssm send-command --document-name "AWS-RunShellScript" --instance-ids $DEV_INSTANCE_ID --region $AWS_DEFAULT_REGION --parameters commands='["cd /home/ubuntu/streamlit","(if [ -d doczy.ai ]; then cd doczy.ai && git fetch && git pull; else git clone git@bitbucket.org:aarete/doczy.ai.git; fi)"]' + trigger: manual # uncomment after testing pipelines: branches: @@ -41,7 +58,14 @@ pipelines: changesets: includePaths: - snowflake/**/* - uat: + - step: + <<: *streamlit_deploy + condition: + changesets: + includePaths: + - streamlit/**/* + # UAT and PROD pipelines will be updated with Streamlit steps once it is tested and ready + UAT: - step: <<: *terraform_plan_apply condition: @@ -52,7 +76,7 @@ pipelines: condition: paths: - snowflake/**/*.sql - prod: + PROD: - step: <<: *terraform_plan_apply condition: From 4f95eece39e6d0e07ba9f7496e87be0234bf6c42 Mon Sep 17 00:00:00 2001 From: Umang Mistry Date: Wed, 6 Mar 2024 16:10:42 -0600 Subject: [PATCH 2/2] Added airflow cicd + reuseable AWS auth step --- airflow/{ => dags}/Qa_Dag.py | 0 airflow/dags/cicd_test_dag.py | 151 +++++++++++++++++++++ airflow/{ => dags}/training_results_dag.py | 0 bitbucket-pipelines.yml | 31 ++++- 4 files changed, 180 insertions(+), 2 deletions(-) rename airflow/{ => dags}/Qa_Dag.py (100%) create mode 100644 airflow/dags/cicd_test_dag.py rename airflow/{ => dags}/training_results_dag.py (100%) diff --git a/airflow/Qa_Dag.py b/airflow/dags/Qa_Dag.py similarity index 100% rename from airflow/Qa_Dag.py rename to airflow/dags/Qa_Dag.py diff --git a/airflow/dags/cicd_test_dag.py b/airflow/dags/cicd_test_dag.py new file mode 100644 index 0000000..86ab6ce --- /dev/null +++ b/airflow/dags/cicd_test_dag.py @@ -0,0 +1,151 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + + +# Mandatory imports +from __future__ import annotations +import os +from datetime import datetime +import logging +from airflow import DAG +from airflow.operators.empty import EmptyOperator +from airflow.utils.trigger_rule import TriggerRule +from airflow.models import Variable + +# SNOWFLAKE only imports +from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator + +# PYTHON only imports +from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook +from airflow.operators.python import PythonOperator + + +# Configuring basic logging for INFO / ERROR in our scripts + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Recommended to use this in the beginning of the script + +SNOWFLAKE_CONN_ID = "doczy_dev_snowflake" # Specific for every database +DAG_ID = "cicd_testing_dag" # Must be unique for every dag +DATABASE="DOCZY_DEV" + +''' +Tags will be used to filter DAGs on the Airflow UI. + +Guidelines for TAGs: +"prod" - for dataload scripts that have been tested +"dev" - for dataloads for DEV databases / scripts in dev +"db_name" - same naming convention as Snowflake databses for clients / projects +"medical" / "pharmacy" - based on the scenario +"etl" / "adhoc" - based on the nature of the dataload + +''' +TAGS=["adhoc","dev"] # MANDATORY + + +''' +TRIGGER RULES for a task: + +all_success: (default) all parents have succeeded +all_failed: all parents are in a failed or upstream_failed state +all_done: all parents are done with their execution +one_failed: fires as soon as at least one parent has failed, it does not wait for all parents to be done +one_success: fires as soon as at least one parent succeeds, it does not wait for all parents to be done +none_failed: all parents have not failed (failed or upstream_failed) i.e. all parents have succeeded or been skipped +none_skipped: no parent is in a skipped state, i.e. all parents are in a success, failed, or upstream_failed state +dummy: dependencies are just for show, trigger at will +''' +ALL_SUCCESS = 'all_success' +ALL_FAILED = 'all_failed' +ALL_DONE = 'all_done' +ONE_SUCCESS = 'one_success' +ONE_FAILED = 'one_failed' + + + +dag = DAG( + # These args will get passed on to each operator + # You can override them on a per-task basis during operator initialization + + # 'queue': 'bash_queue', + # 'pool': 'backfill', + # 'priority_weight': 10, + # 'end_date': datetime(2016, 1, 1), + # 'wait_for_downstream': False, + # 'sla': timedelta(hours=2), + # 'execution_timeout': timedelta(seconds=300), + # 'on_failure_callback': some_function, + # 'on_success_callback': some_other_function, + # 'on_retry_callback': another_function, + # 'sla_miss_callback': yet_another_function, + # 'trigger_rule': 'all_success' + + DAG_ID, # Mandatory for every dag + start_date=datetime(2022, 1, 1), # Must be in the past + # Can pass snowflake conn id here instead of passing it to every task + # It is needed when using Snowflake operator + default_args={"snowflake_conn_id": SNOWFLAKE_CONN_ID, 'retries': 0}, + tags=TAGS, + catchup=False, # True will run the dag on the specified frequency for backdated DAGs. Not applicate in out workflow + schedule=None # If there is no fixed schedule, then always pass None explicitly + +) + + +def python_op_eg(schema_name,table_name): + + # Snowflake hook is used to fetch connection and cursor + dwh_hook = SnowflakeHook(snowflake_conn_id=SNOWFLAKE_CONN_ID) + conn = dwh_hook.get_conn() + curr = conn.cursor() + query_output = curr.execute(f"SELECT * from {DATABASE}.{schema_name}.{table_name}") + + # Alternative way to fetch query result + + # result = dwh_hook.get_first(f"select max(audit_sid) from {DATABASE}.stg.CLAIM_MED_STAGING") + # max_audit_sid = result[0] + + logging.info(f"PYTHON OPERATOR OUTPUT = {query_output.fetchall()}") + conn.close + + +# Best practice to have an empty start at the beginning and end +begin_job = EmptyOperator(task_id='Begin') + +python_task = PythonOperator(task_id='get_info_using_python_op', # Task ID has to be unique only inside a dags + python_callable=python_op_eg, + dag=dag, + op_kwargs={'table_name':'DIM_AUDIT', 'schema_name': 'STG'} # Variables can be passed to the python function using op_kwargs + ) + +# Snowflake operator does not offer the option to display query results +# We can use this for executing stored procedures. +sf_task = SnowflakeOperator( + task_id="get_info_using_sf_op", + sql=f"select * from {DATABASE}.STG.DIM_AUDIT", + dag=dag +) + + + +end_job = EmptyOperator(task_id='End') + +# EXECUTING TASKS +begin_job >> python_task >> sf_task >> end_job \ No newline at end of file diff --git a/airflow/training_results_dag.py b/airflow/dags/training_results_dag.py similarity index 100% rename from airflow/training_results_dag.py rename to airflow/dags/training_results_dag.py diff --git a/bitbucket-pipelines.yml b/bitbucket-pipelines.yml index e8132ab..69bb66f 100644 --- a/bitbucket-pipelines.yml +++ b/bitbucket-pipelines.yml @@ -1,10 +1,21 @@ definitions: steps: + - step: &aws_auth_script + name: "Authenticate with AWS" + script: + - export AWS_ROLE_ARN=arn:aws:iam::$AWS_ACCOUNT_NO:role/$OIDC_ROLE + - export AWS_WEB_IDENTITY_TOKEN_FILE=$(pwd)/web-identity-token + - echo $BITBUCKET_STEP_OIDC_TOKEN > $(pwd)/web-identity-token + - export STS_OUTPUT=$(aws sts assume-role-with-web-identity --role-arn $AWS_ROLE_ARN --role-session-name BitbucketPipeline --web-identity-token "$BITBUCKET_STEP_OIDC_TOKEN" --duration-seconds 3600) + - export AWS_ACCESS_KEY_ID=$(echo $STS_OUTPUT | jq -r '.Credentials.AccessKeyId') + - export AWS_SECRET_ACCESS_KEY=$(echo $STS_OUTPUT | jq -r '.Credentials.SecretAccessKey') + - export AWS_SESSION_TOKEN=$(echo $STS_OUTPUT | jq -r '.Credentials.SessionToken') + - aws sts get-caller-identity - step: &terraform_plan_apply name: "Plan and Apply Terraform" image: zenika/terraform-aws-cli:latest script: - - terraform --version + - terraform --version # terraform steps here trigger: automatic # uncomment after testing # trigger: manual @@ -29,7 +40,7 @@ definitions: name: "Deploy streamlit to EC2" image: python:3.8 script: - - python -m pip install --upgrade pip + - python -m pip install --upgrade pip - apt-get update && apt-get install -y jq git - pip install awscli - export AWS_ROLE_ARN=arn:aws:iam::$AWS_ACCOUNT_NO:role/$OIDC_ROLE @@ -42,6 +53,16 @@ definitions: - aws sts get-caller-identity - aws ssm send-command --document-name "AWS-RunShellScript" --instance-ids $DEV_INSTANCE_ID --region $AWS_DEFAULT_REGION --parameters commands='["cd /home/ubuntu/streamlit","(if [ -d doczy.ai ]; then cd doczy.ai && git fetch && git pull; else git clone git@bitbucket.org:aarete/doczy.ai.git; fi)"]' trigger: manual # uncomment after testing + - step: &airflow_dags_deploy + name: "Deploy Airflow DAGs to S3 bucket" + image: python:3.8 + script: + - python -m pip install --upgrade pip + - apt-get update && apt-get install -y jq git + - pip install awscli + - *aws_auth_script + - aws s3 sync /airflow/dags s3://doczy-dev-infra-mwaa-resources/dags + trigger: automatic # uncomment after testing pipelines: branches: @@ -64,6 +85,12 @@ pipelines: changesets: includePaths: - streamlit/**/* + - step: + <<: *airflow_dags_deploy + condition: + changesets: + includePaths: + - airflow/dags/* # UAT and PROD pipelines will be updated with Streamlit steps once it is tested and ready UAT: - step: