Ejecución de los pipeline

PASOS PARA EJECUTAR EL PIPELINE

1.- PREREQUISITOS.

Es necesario contar con los siguientes software instalados y contar con accesos a las siguientes herramientas que a continuación se mencionan.

Vs Code - Para el desarrollo en código de los pipelines
Git - Para el control de versionado de código
python 3.X - Lenguaje en el que se desarrollan los pipelines
OpenVPN - Para tener acceso al repositorio en Github
Cuenta en Github - Tener una cuenta en Github para ver el proyecto en la plataforma
CLI gcloud - Para la conexión con las herramientas en la nube de Google
Accesos a herramientas de gcloud como:
- Bigquery proyectos ( spin-datalake para dev, qa, trusted)
- Dataflow ( spin-compute-dev & spin-compute-qa)
- Cloud Storage (spin-compute-dev & spin-compute-qa)
Librerías de Python:
- Apache beam - pip install "apache-beam[gcp,interactive,dataframe]"
- Pandas - pip install pandas
- Numpy - pip install numpy

2.- CLONA EL REPOSITORIO

Una vez teniendo instalada la VPN, cuenta en Github y con accesos al proyecto, se clona el repositorio dentro de una carpeta de preferencia en el ambiente local, esto haciéndolo por vía SSH (para más información).

En la terminal ubicado en la ruta deseada para clonar el repositorio, ejecuta:

git clone https://github.com/github/<repositorio>

Con VSC abre la carpeta del proyecto para visualizar su estructura.

https://github.com/fintechdigitalventure/spin-data-mirrorstrategy

3.- CONSIDERACIONES

Para correr el pipeline de forma local se deben tener las siguientes consideraciones

Tener instalado previamente el CLI de gcloud ya que es la manera en que nos autenticaremos con las credenciales de gcp desde la terminal de nuestro equipo local
Estar conectado a la VPN de SPIN
Tener un ambiente en python (ya sea con anaconda o pyenv) y tener instalado el SDK para trabajar con apache beam, para eso ejecutar un pip en la terminal con lo siguiente:
- pip install "apache-beam[gcp,interactive,dataframe]"
Para evitar que falte alguna dependencia, estos serian los requirements.txt para que se pueda ejecutar correctamente de forma local. Algunas librerías se instalan de forma automática al instalar el sdk de apache beam.

Una vez teniendo instalado el CLI de gcloud y teniendo nuestro ambiente de python preparado con las librerías necesarias, se deben ejecutar los siguientes comandos en la terminal para el login en gcp.
- # esto nos va a inicializar la configuración para loguernos gcloud init
Nos pedirá login con una cuenta, seleccionamos Log in with a new account y se nos abrirá una ventana del explorador donde podremos ingresar las credenciales de nuestra cuenta

Después nos pedirá elegir el proyecto de gcp con el que queremos trabajar, es importante que las pruebas que estemos corriendo en local con los permisos de service accounts, dataflow, sean las correctas con el proyecto que elegiremos, ejemplo: no elegir un proyecto de QA corriendo pruebas en DEV porque posiblemente marque permisos insuficientes. escribimos el numero del proyecto [spin-compute-dev, spin-compute-qa]

Por ultimo nos pedirá set de la región, como el proyecto se encuentra en US elegimos us-east4

Configuración de VSC:

{
    "version": "0.2.0",
    "configurations": [
        
        {
            "name": "Python: Current File",
            "type": "python",
            "request": "launch",
            "program": "${file}",
            "console": "integratedTerminal",
            "cwd": "${workspaceFolder}",
            "env": {
                "PYTHONPATH":"${PYTHONPATH}:/Users/paloit2023/Documents/Proyectos/MIRROR/spin-mirror-strategy-dev"
            }
        }
    ]
}

Colocar en PYTHONPATH : la ubicación de la carpeta del proyecto.

Con la anterior configuración ya podremos correr el pipeline de manera local, esto permitirá realizar pruebas del funcionamiento del código desarrollado.

4.- CONFIGURACIÓN DE AMBIENTE

Bucket templates

Creación de carpetas dentro de spin-dev-code-data:

mirror
- dataflow
  - temp
  - tmp
- staging
- temp
- template

En la siguiente ruta spin-dev-code-data/sql/dml subir los archivos:

delete_landing_mirror.txt
delete_sync_mirror.txt
delete_duplicate_mirror.txt

del la ruta templates/sql/dml/ del repositorio

Bucket de airflow

En la siguiente ruta us-central1-orquestado-10f088be-bucket/dags/vars subir el archivo:

dml_mirror.py

de la ruta composer/vars/dml_mirror.py del repositorio

5. CREACIÓN DE TEMPLATES

Para la creación de los templates se deberá seguir lo siguiente:

Template de parse: Se migran datos de Landing a Raw

Bucket: spin-dev-code-data/mirror

En la carpeta migrate del proyecto existen las carpetas:

daily: hace la ejecución diaria de un append de datos
historic: hace la ejecución de datos por primera vez

En cada carpeta existen 3 archivos con terminación parse.py.

abrir el archivo del cual se quiere generar un template y colocar el ambiente en el cual se ejecutara (dev, qas, prd)

Se debe de ejecutar el archivo en modo depuración:
Se generara el siguiente comando: python3 -m templates.template_parse --project spin-compute-dev --runner DataflowRunner --staging-location gs://spin-dev-code-data/mirror/staging --temp-location gs://spin-dev-code-data/mirror/temp --template_location gs://spin-dev-code-data/mirror/template/spin_daily_mongo_mirror_account_parse_dev --region us-east1

El comando se deberá ejecutar en la terminal en la ruta del proyecto una ves activado el ambiente de python previamente configurado:

El comando anterior creará el template en la siguiente ruta: spin-dev-code-data/mirror/template

Repetir el paso para la creación de todos los templates de la carpeta daily tanto para los archivos de terminación parse.py como para los de transformation.py:

Template de transform: se migran datos de Raw a Trusted

Bucket: spin-dev-code-data/mirror

En la carpeta migrate del proyecto existen las carpetas:

daily: hace la ejecución diaria de un append de datos
historic: hace la ejecución de datos por primera vez

En cada carpeta existen 3 archivos con terminación transformation.py

Nota: Así mismo repetir el proceso con los templates de la carpeta historic

Se generaran los siguientes archivos:

spin_daily_mongo_mirror_account_parse_dev

spin_daily_mongo_mirror_account_transform_dev

spin_daily_mongo_mirror_balance_parse_dev

spin_daily_mongo_mirror_balance_transform_dev

spin_daily_mongo_mirror_card_parse_dev

spin_daily_mongo_mirror_card_transform_dev

spin_historic_mongo_mirror_account_parse_dev

spin_historic_mongo_mirror_account_transform_dev

spin_historic_mongo_mirror_balance_parse_dev

spin_historic_mongo_mirror_balance_transform_dev

spin_historic_mongo_mirror_card_parse_dev

spin_historic_mongo_mirror_card_transform_dev

cambiando la terminación dev por el ambiente que se este ejecutando (qas,prd)

6.- PRUEBA DE EJECUCIÓN LOCAL. (No necesario para despliegue en ambientes)

Seleccionar proceso a correr.

Una vez se termine la ejecución de forma satisfactoria veremos el siguiente msn: {'status': 'SUCCESS', 'message': 'Execution successfully', 'output': {}}

6.-EVIDENCIA DE FUNCIONAMIENTO EN GCP

Airflow

PROCESO HISTÓRICO:

Se cuenta con 3 DAG históricos que cargan la data de landing a raw y de raw a trusted, este proceso se ejecuta solo una vez de manera manual y trae todos los datos existentes en landing al momento de la ejecución.

Los dag son:

historic_mongo_mirror_account
historic_mongo_mirror_balance
historic_mongo_mirror_card

PROCESO DIARIO:

Se cuenta con 3 DAG diario que cargan la data de raw a trusted, este proceso se ejecuta solo una vez al día de manera automática y trae todos los datos existentes landing a raw que no tengan _fivetran_deleted = true y de raw a trusted, este proceso solo trae los registros nuevos, es decir, aquellos que no hayan sido insertados en el proceso histórico o en una ejecución de un proceso daily.