A Complete Step-by-Step Guide to Building Data Pipeline for ETL Operation Using Modern Data Engineering Tools

To do:

 In this tutorial, we will be covering different phases we completed in building our data pipeline:

Writing Python script to create a Fivetran connector connecting datasets from Google Sheets to data from our Snowflake warehouse.

Configuring Snowflake with Data Built tools (DBT).

Writing SQL transformation in DBT and deploying models back in our warehouse.

Connecting our warehouse with Looker Studio and creating interactive visualizations.

About the dataset:

We download an open-source dataset from Kaggle of sales data. It contained categories of each product with their cost price and price each for which he sold each item, turnover, and quantity ordered for each product in their category.

Each order is identified with its unique order ID and the ordering date. The total dataset contained 185951 rows including headers.

Step 01: Creating Fivetran Connector using Python.

Fivetran provides us REST framework within Python to send API calls to perform tasks. This can be found in their official API documentation. In our code, we used POST calls with two predefined endpoints. First, we used the ‘connector’ endpoint with configures defined in our payload variable. Then we used a sync connector with that specified endpoint to sync our data.

It should be noted that the proper API and other credentials are required which will be present in the Fivetran user profile. Similarly, we configured our warehouse connection while creating the Fivetran user profile to get specific credentials for proper connections which will be further used in our code.

Code:

## Importing libraries

import requests

from requests.auth import HTTPBasicAuth

import json

import colorama

from colorama import Fore, Back, Style

## Account Credentials | This will be obtained from the API tab within the Fivetran Profile Account

api_key = ''

api_secret = ''

a = HTTPBasicAuth(api_key, api_secret)

## Function for API calls

def atlas(method, endpoint, payload=None):

base_url = 'https://api.fivetran.com/v1'

h = {

'Authorization': f'Bearer {api_key}:{api_secret}'

}

url = f'{base_url}/{endpoint}'

try:

if method == 'GET':

response = requests.get(url, headers=h, auth=a)

elif method == 'POST':

response = requests.post(url, headers=h, json=payload, auth=a)

elif method == 'PATCH':

response = requests.patch(url, headers=h, json=payload, auth=a)

elif method == 'DELETE':

response = requests.delete(url, headers=h, auth=a)

else:

raise ValueError('Invalid request method.')

response.raise_for_status()  # Raise exception for 4xx or 5xx responses

return response.json()

except requests.exceptions.RequestException as e:

print(f'Request failed: {e}')

return None

## Creating our connector

method = 'POST'

endpoint = 'connectors'

payload = {

"service": "google_sheets",

"group_id": "",    ## Paste your group_id from the Fivetran Profile Data Tab

"config": {

"schema": "test_sheet_schema",

"table": "test_sheet_table",

"sheet_id": "",     ## Paste your google sheet link

"named_range": "test_sheet_range",    ## Replace you selected named range within the google sheets

"auth_type": "ServiceAccount"

}

}

response = atlas(method, endpoint, payload)

print(response)

## Sync data from sheet

method = 'POST'

connector_id = ''  ## Place your connector ID here from you Fivetran profile Setup tab

endpoint = f'connectors/{connector_id}/sync'

payload = {

"force": False

}

response = atlas(method, endpoint, payload)

print(response)