Dagster

Data Orchestration with python

pipeline

Se ha construido una aplicación donde recopile información de un archivo excel y este sea leido, transformado y cargado a la base de datos local.

Ejecución

Necesario python 3.7+

Para el el correcto funcionamiento de daemon en windows es necesario acceder a las variables de entorno del sistema y declarar una variable llamada DAGSTER_HOME con la ruta en donde se encuentra la carpeta dagster_home*

Ejemplo

 #Ejemplo: \c\Users\Usuario\Documents\data-engineering\pipelines\dagster\dagster_home

Por ahora se puede ejecutar en un entorno local añadiendo un archivo en el directorio raíz con nombre ".env" con la información como se muestra a continuación.

DAGSTER_HOME = path\dagster_home        #Ejemplo: \c\Users\Usuario\Documents\data-engineering\pipelines\dagster\dagster_home
DAGSTER_PG_USERNAME = usuario
DAGSTER_PG_PASSWORD = contraseña
DAGSTER_PG_HOST = host
DAGSTER_PG_DB = nombre_bd
DAGSTER_PG_PORT = puerto

Y tambien es necesario actualizar la información de la base de datos en el archivo dagster.yaml situada en la carpeta en donde se declaró la variable de entorno dagster_home. Cambiar los siguientes valores:

postgres_db:
      username: postgres
      password: password
      hostname: localhost
      db_name: dagster
      port: 5432

Para ejecutar el script es necesario situarse en la carpeta raíz del proyecto y ejecutar la siguiente sentencia:

python ETL.py

O a través de dagit (IU) hay que ejecutar la siguiente sentencia:

dagit -f ETL.py

Y dentro del servidor en la pestaña de launchpad presionar el boton launch run y llevará a cabo el pipeline.

Schedule

Importante

Para el schedule de este pipeline es necesario correr el siguiente comando en una terminal y mantenerlo activo, el cual leera el archivo dagster.yaml

dagster-daemon run

Una vez activado el daemon, en otra terminal ejecutamos el comando:

dagit -f ETL.py

Dentro de la IU de dagit ingresamos revisamos el status y nos dirigiremos a schedules, una vez estemos dentro de la ruta, activamos el schedule update_warehouse_schedule presionando el toogle

Final

Ya que se activo esta funcionalidad, el pipeline se ejecutará cada 24hrs a las 2:00AM.

Muestras