Hey world, the concept of ETL are far from new, but nowadays it is widely used in the industry. ETL stands for Extract, Transform, and Load. Okay, but what does that mean? The easiest way to understand how ETL works is to understand what happens in each step of the process. Let's dive into it.
Extract
During the extraction, raw data is moved from a structured or unstructured data pool to a staging data repository.
Transform
The data source might have a different structure than the target destination, we'll transform the data from the source schema to the destination schema.
Load
In this phase, we'll then load the transformed data into the data warehouse.
A disadvantage of the ETL approach is that the transformation stage can take a long time. An alternative approach is extract, load, and transform (ELT). In ELT, the data is immediately extracted and loaded into a large data repository, such as Azure Data Lake Storage. We can begin transforming the data as soon as the load is complete.
Hands on
In this first part I will show how to create an ELT. We'll extract data from a Public API called IntegraSUS regarding Covid-19 data, and load it on Azure Data Lake Storage. So, this ELT will be containerized on Azure Container Registry (ACR), and we will use Azure Kubernetes Service (AKS) to schedule our job on K8s cluster to run daily.
In the second part of this project, we will integrate the Azure Data Lake with Apache Spark on Azure Databricks to perform a small transformation on top of the files sent to the Data Lake and then we will store the result of the transformation in a Data Warehouse.
We will learn how to:
- Create a Python ELT and load into Azure Data Lake;
- Create an Azure Container Registry and push images into it;
- Create an Azure Kubernetes Service;
- Deploy CronJob into Azure Kubernetes Cluster;
- Integrate Azure Data Lake with Apache Spark on Databricks;
- Transform data using PySpark on Azure Databricks
- Load new data into Data Warehouse.
The project code is available here: github repository
1. Create a Python ELT
In the extraction phase we will get data from a Public API about Fortaleza/Ceará/Brazil Covid-19 data, and store the data into a json file. After that, we will load it into Azure Data Lake. You can see the project code below.
#!/usr/local/bin/python
import os
import sys
import yaml
import json
import logging
import requests
from datetime import datetime, timedelta
from azure.storage.filedatalake import DataLakeServiceClient
logging.basicConfig(stream=sys.stdout,
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
date = datetime.today() - timedelta(days=2)
previous_date = f"{date.year}-{date.month}-{date.day}"
def extract():
logger.info('EXTRACTING...')
#extract data from API
url = 'https://indicadores.integrasus.saude.ce.gov.br/api/casos-coronavirus?dataInicio='+ previous_date +'&dataFim=' + previous_date
req = requests.get(url)
data = req.json()
if data:
with open(f'covid-data-{previous_date}.json', 'w', encoding='utf-8') as file:
json.dump(data, file, ensure_ascii=False, indent=4)
else:
logger.info('THERE IS NOT DATA')
def initialize_storage_account(storage_account_name, storage_account_key):
try:
global service_client
service_client = DataLakeServiceClient(account_url="https://"+ storage_account_name +".dfs.core.windows.net", credential=storage_account_key)
except Exception as e:
logger.info('EXCEPTION...')
logger.info(e)
def create_file_system(container_name):
try:
global file_system_client
logger.info('CREATING A CONTAINER NAMED AZ-COVID-DATA')
file_system_client = service_client.create_file_system(file_system=container_name)
except Exception as e:
logger.info('EXCEPTION...')
logger.info(e)
def create_directory():
try:
logger.info('CREATING A DIRECTORY NAMED DIRECTORY-COVID22')
file_system_client.create_directory("directory-covid19")
except Exception as e:
logger.info('EXCEPTION...')
logger.info(e)
def upload_file_to_container_datalake(local_file, container_name):
try:
logger.info('UPLOADING FILE TO AZURE DATA LAKE STORAGE...')
file_system_client = service_client.get_file_system_client(file_system=container_name)
directory_client = file_system_client.get_directory_client("directory-covid19")
file_client = directory_client.get_file_client(f"covid-{previous_date}.json")
with open(local_file, "rb") as data:
file_client.upload_data(data, overwrite=True)
logger.info('UPLOADED TO AZURE DATA LAKE')
except Exception as e:
logger.info('EXCEPTION...')
logger.info(e)
def load_config():
directory = os.path.dirname(os.path.abspath(__file__))
with open(directory + "/config.yaml", "r") as yamlfile:
return yaml.load(yamlfile, Loader=yaml.FullLoader)
if __name__ == "__main__":
extract()
config = load_config()
initialize_storage_account(config["AZURE_DL_STORAGE_ACCOUNT_NAME"], config["AZURE_DL_ACCOUNT_KEY"])
create_file_system(config["AZURE_DL_CONTAINER_NAME"])
upload_file_to_container_datalake(f"covid-data-{previous_date}.json", config["AZURE_DL_CONTAINER_NAME"])
2. Create docker image for our Python ELT
We are going to build a docker image for our ELT job and run it inside the container. So, let's create a Dockerfile, which describes how a docker image is built. You can see a list of instructions below.
FROM python:3.9.12-buster
WORKDIR /usr/src/app
COPY requirements.txt /usr/src/app
RUN pip install -r requirements.txt
COPY config.yaml /usr/src/app
COPY el2datalake.py /usr/src/app
RUN chmod a+x el2datalake.py
CMD ["./el2datalake.py"]
We can build the docker image using docker build command
$ docker build -t el2datalakejob .
Now we should run our ELT inside the container
$ docker run -it el2datalakejob:latest
3. Push docker images to Azure Container Registry
Azure Container Registry handles private Docker container images and allow us to build, store, and manage container images. We are going to deploy an ACR instance and push a docker image to it.
To create an any instance on Azure, we must create a resource group. We create a new resource group with az group create command.
$ az group create --name myResourceGroup --location westeurope
Once we have a resource group, we can create an Azure Container Registry with az acr create command.
$ az acr create \
--resource-group myResourceGroup \
--name azcrjobs \
--sku Basic \
--location westeurope
Let's login on Azure Container Registry
$ az acr login --name azcrjobs
Let us tag the image to the login server azcrjobs.azurecr.io
$ docker tag el2datalakejob \
azcrjobs.azurecr.io/el2datalakejob:v1
Push the Docker image to ACR
$ docker push azcrjobs.azurecr.io/el2datalakejob:v1
Now we have our ELT on Azure Container Registry let's move on to the next step.
4. Create and Deploy CronJobs on Azure Kubernetes Service
Azure Kubernetes Service (AKS) deploy and manage containerized applications more easily with a fully managed Kubernetes service. Let’s create an AKS cluster with az aks create command.
$ az aks create \
--resource-group myResourceGroup \
--name az-aks-jobs \
--node-count 1 \
--attach-acr azcrjobs \
--location westeurope
To connect to the cluster from local machine we use Kubernetes client kubectl, open the terminal to connect to the cluster
$ az aks get-credentials --resource-group myResourceGroup \ --name az-aks-jobs
Let us see our node available on AKS
$ kubectl get nodes
We start from creating a manifest file for our ELT cron job.
apiVersion: batch/v1
kind: CronJob
metadata:
creationTimestamp: null
name: k8sjob
spec:
jobTemplate:
metadata:
creationTimestamp: null
name: k8sjob
spec:
template:
metadata:
creationTimestamp: null
spec:
containers:
- image: azcrjobs.azurecr.io/el2datalakejob:v1
imagePullPolicy: IfNotPresent
name: k8sjob
resources: {}
restartPolicy: OnFailure
schedule: '55 23 * * *'
status: {}
Above on our manifest file we defined the crontab expression used as a schedule for our job, and is scheduled to run everyday at 23:55. We put the name of the docker image to be pulled from container registry attached to cluster.
To deploy our job, we will use the kubectl apply command.
kubectl apply -f job.yml
We can view some details about the job with
kubectl get cronjobs
To retrieve cron job logs from Kubernetes, we can use kubectl logs command, but first we must get the pod name.
$ kubectl get pods
NAME READY STATUS RESTARTS AGE
k8sjob-27513350--1-xnj8x 0/1 Completed 0 4m2s
Retrieve cron job logs from Kubernetes
$ kubectl logs k8sjob-27513350--1-xnj8x
Conclusion
Finally, we have the first stage of our project completed. Now we have Covid data on Azure Data Lake. For the next step, we will read this file from Azure Data Lake and perform a little processing of this data using Apache Spark on Azure Databricks, and we will be able to make the result of this processing available in a Data Warehouse.
Top comments (0)