ELT Data Pipeline with Kubernetes CronJob, Azure Data Lake, Azure Databricks (Part 1)
Tags: etl, dataengineering, python, azure
Published: 2022-04-24 14:35:29

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](https://github.com/rubnsbarbosa/elt2datalake)

**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.

```python
#!/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.

```python
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.