Building an ETL pipeline with Apache Airflow and Visualizing AWS Redshift data using Microsoft Power BI

Check the article here: Building an ETL data pipeline with Apache Airflow and Visualizing AWS Redshift data using Microsoft Power BI

Architecture - Uber expenses tracking

What are the data sources?

Uber Rides receipt example

Uber Eats receipt example

Data modelling

Infrastructure as Code (IaC) in AWS

Below we list the different steps and the things carried out in this file:

• Install git-bash for windows, once installed , open git bash and download this repository, this will download the dags folder and the docker-compose.yaml file, and other files needed.

ramse@DESKTOP-K6K6E5A MINGW64 /c
$ git clone https://github.com/Wittline/uber-expenses-tracking.git

• Second, create a new User in AWS with AdministratorAccess* and get your security credentials
• Go to this url: AWS CLI and configure your AWS Credentials in your local machine
• Setup local environment with Google Colab:
  • Go to CMD and put:

    C:\>jupyter notebook --NotebookApp.allow_origin='https://colab.research.google.com' --port=8888 --NotebookApp.port_retries=0
    

  • After the execution of the last command, copy the localhost url, you will need it for colab
  • Go to Google Colab
  • Create a new Notebook
  • Go to -> Connect -> "Connect to local runtime" -> Paste the url copied from the last step and put it in Backend URL -> connect
  • Upload the file AWS-IAC-IAM-EC2-S3-Redshift.ipynb, and use it into your colab local env:
    • Create the required S3 buckets
      • uber-tracking-expenses-bucket-s3
      • airflow-runs-receipts
    • Move Uber receipts from your local computer to S3
    • Loading params from the dwh.cfg file
    • Creating clients for IAM, EC2 and Redshift cluster
    • Creating the IAM Role that makes Redshift able to access to S3 buckets ReadOnly
    • Creating Redshift cluster
    • Check cluster details until status show Available
    • Showing Redshift cluster endpoint and role ARN
    • Incoming TCP port to access to the cluster ednpoint
    • Checking the connection to the Redshift cluster
    • Cleaning and deleting the resources

Content of AWS-IAC-IAM-EC2-S3-Redshift.ipynb

Libraries

import pandas as pd
import glob
import os
import boto3
import json
import configparser
from botocore.exceptions import ClientError
import psycopg2

Cloning repository, Buckets creation, folders and uploading the local files to S3

def bucket_s3_exists(b):
    s3 = boto3.resource('s3')
    return s3.Bucket(b) in s3.buckets.all()

def create_s3_bucket(b, folders):
    client = boto3.client('s3')
    client.create_bucket(Bucket=b, CreateBucketConfiguration={'LocationConstraint': 'us-east-2'})
    if folders is not '':
        fls = folders.split(',')
        for f in fls:
            client.put_object(Bucket= b, Body='', Key=f + '/')

def upload_files_to_s3(file_name, b, folder, object_name, args=None):
    
    client = boto3.client('s3')

    if object_name is None:
        object_name = folder + "/{fname}".format(fname= os.path.basename(file_name)) 

    response = client.upload_file(file_name, b, object_name , ExtraArgs = args)

    return response


ACLargs = {'ACL':'authenticated-read' }
bucket_names = {'uber-tracking-expenses-bucket-s3': 'unprocessed_receipts', 'airflow-runs-receipts':'eats,rides'}

print('Creating the S3 buckets...')
for k in bucket_names:
    if not bucket_s3_exists(k):
       create_s3_bucket(k, bucket_names[k])    

print('S3 buckets created')

print('Uploading the local receipts files to uber-tracking-expenses-bucket-s3 AWS S3 bucket...')
files = glob.glob("localpath/receipts/*")

for file in files:
    print("Uploading file:", file)
    print(upload_files_to_s3(file, 'uber-tracking-expenses-bucket-s3', 'unprocessed_receipts', None, ACLargs))


print('Files uploaded to uber-tracking-expenses-bucket-s3 AWS S3 bucket')
 

Loading all the Params from the dwh.cfg file

config = configparser.ConfigParser()
config.read_file(open('/Uber-expenses-tracking/IAC/dwh.cfg'))

KEY                    = config.get('AWS','KEY')
SECRET                 = config.get('AWS','SECRET')

DWH_CLUSTER_TYPE       = config.get("DWH","DWH_CLUSTER_TYPE")
DWH_NUM_NODES          = config.get("DWH","DWH_NUM_NODES")
DWH_NODE_TYPE          = config.get("DWH","DWH_NODE_TYPE")

DWH_CLUSTER_IDENTIFIER = config.get("DWH","DWH_CLUSTER_IDENTIFIER")
DWH_DB                 = config.get("DWH","DWH_DB")
DWH_DB_USER            = config.get("DWH","DWH_DB_USER")
DWH_DB_PASSWORD        = config.get("DWH","DWH_DB_PASSWORD")
DWH_PORT               = config.get("DWH","DWH_PORT")

DWH_IAM_ROLE_NAME      = config.get("DWH", "DWH_IAM_ROLE_NAME")

(DWH_DB_USER, DWH_DB_PASSWORD, DWH_DB)

pd.DataFrame({"Param":
                 ["DWH_CLUSTER_TYPE", "DWH_NUM_NODES", "DWH_NODE_TYPE", "DWH_CLUSTER_IDENTIFIER", 
                  "DWH_DB", "DWH_DB_USER", "DWH_DB_PASSWORD", "DWH_PORT", "DWH_IAM_ROLE_NAME"],
             "Value":
                 [DWH_CLUSTER_TYPE, DWH_NUM_NODES, DWH_NODE_TYPE, DWH_CLUSTER_IDENTIFIER, 
                 DWH_DB, DWH_DB_USER, DWH_DB_PASSWORD, DWH_PORT, DWH_IAM_ROLE_NAME]
            })

Creating clients for IAM, EC2 and Redshift ressources

ec2 = boto3.resource('ec2',
                      region_name="us-east-2",
                      aws_access_key_id=KEY,
                      aws_secret_access_key=SECRET
                   )

iam = boto3.client('iam',aws_access_key_id=KEY,
                    aws_secret_access_key=SECRET,
                    region_name='us-east-2'
                 )

redshift = boto3.client('redshift',
                      region_name="us-east-2",
                      aws_access_key_id=KEY,
                      aws_secret_access_key=SECRET
                      )

Creating the IAM Role that makes Redshift able to access S3 buckets (ReadOnly)

try:
   print("Creating new IAM Role") 
   dwhRole = iam.create_role(
       Path='/',
       RoleName=DWH_IAM_ROLE_NAME,
       Description = "Allows Redshift clusters to call AWS services on your behalf.",
       AssumeRolePolicyDocument=json.dumps(
           {'Statement': [{'Action': 'sts:AssumeRole',
              'Effect': 'Allow',
              'Principal': {'Service': 'redshift.amazonaws.com'}}],
            'Version': '2012-10-17'})
   )    
except Exception as e:
   print(e)
   
   
print("Attaching Policy")

iam.attach_role_policy(RoleName=DWH_IAM_ROLE_NAME,
                      PolicyArn="arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess"
                     )['ResponseMetadata']['HTTPStatusCode']

print("Get the IAM role ARN")
roleArn = iam.get_role(RoleName=DWH_IAM_ROLE_NAME)['Role']['Arn']

print(roleArn)

Creating Redshift Cluster

try:
   response = redshift.create_cluster(        
       
       ClusterType=DWH_CLUSTER_TYPE,
       NodeType=DWH_NODE_TYPE,
       NumberOfNodes=int(DWH_NUM_NODES),


       DBName=DWH_DB,
       ClusterIdentifier=DWH_CLUSTER_IDENTIFIER,
       MasterUsername=DWH_DB_USER,
       MasterUserPassword=DWH_DB_PASSWORD,
       
  
       IamRoles=[roleArn]  
   )
except Exception as e:
   print(e)

Redshift Cluster Details (Run ths piece of code several times until status show Available)

def prettyRedshiftProps(props):
   pd.set_option('display.max_colwidth', -1)
   keysToShow = ["ClusterIdentifier", "NodeType", "ClusterStatus", "MasterUsername", 
                 "DBName", "Endpoint", "NumberOfNodes", 'VpcId']
   x = [(k, v) for k,v in props.items() if k in keysToShow]
   return pd.DataFrame(data=x, columns=["Key", "Value"])

myClusterProps = redshift.describe_clusters(ClusterIdentifier=DWH_CLUSTER_IDENTIFIER)['Clusters'][0]
prettyRedshiftProps(myClusterProps)

Redshift Cluster endpoint and role ARN

DWH_ENDPOINT = myClusterProps['Endpoint']['Address']
DWH_ROLE_ARN = myClusterProps['IamRoles'][0]['IamRoleArn']
print("DWH_ENDPOINT :: ", DWH_ENDPOINT)
print("DWH_ROLE_ARN :: ", DWH_ROLE_ARN)

Incoming TCP port to access to the cluster endpoint

try:
   vpc = ec2.Vpc(id=myClusterProps['VpcId'])
   defaultSg = list(vpc.security_groups.all())[0]
   print(defaultSg)
   defaultSg.authorize_ingress(
       GroupName=defaultSg.group_name,
       CidrIp='0.0.0.0/0',
       IpProtocol='TCP',
       FromPort=int(DWH_PORT),
       ToPort=int(DWH_PORT)
   )
except Exception as e:
   print(e)

Checking the connection to the redshift cluster

conn_string="postgresql://{}:{}@{}:{}/{}".format(DWH_DB_USER, DWH_DB_