Bash Operator

Clase 12 de 29 • Curso de Fundamentos de Apache Airflow

Clase anteriorSiguiente clase

Shirley Tatiana Pitta Picon

student•

Camilo Andrés Rodriguez Higuera

student•

El resultado

Randy José Agustín Montenegro Socha

student•

mi resultado:

josefabian cardonahernandez

student•

tuve un problema como se muestra en la figura, usando el mismo codigo, si lo hacia corrrer manual funcionaba pero a veces persiste el error, si alguien puede explicarlo, lo agradezco.

Jose Daniel Velasquez H

company_admin•

Brayan coveñas

student•

Consulta regularmente es por comodidad o porque estoy acostumbrado pero saben como podria hacer el autocompletado con airflow?

Jose Alejandro Rivillas

student•

No me aparece el botón de logs para la tarea

Jose Alejandro Rivillas

student•

Tenía que corregir la librería desde donde se importaba

Gino G. Viloria

student•

Daniel Olave

student•

Hice un sencillo hello world jiji

Mario Alexander Vargas Celis

student•

El BashOperator en Apache Airflow permite ejecutar comandos de Bash en un flujo de trabajo. Es útil para realizar tareas como mover archivos, ejecutar scripts de shell, o interactuar con herramientas del sistema operativo desde un DAG.

Estructura Básica del BashOperator

Aquí tienes un ejemplo básico de cómo usar el BashOperator:

from airflow import DAG from airflow.operators.bash import BashOperator from datetime import datetime

# Definir el DAG with DAG( dag_id="bash_operator_example", description="Ejemplo de uso del BashOperator", start_date=datetime(2024, 11, 28), # Fecha en el pasado schedule_interval="@daily", # Se ejecuta diariamente catchup=False, # No ejecuta tareas pendientes de fechas pasadas ) as dag:

# Definir una tarea usando BashOperator t1 = BashOperator( task_id="print_date", bash_command="date", # Comando de Bash que imprime la fecha actual )

# Otra tarea para ejecutar un script t2 = BashOperator( task_id="run_script", bash_command="echo 'Ejecutando mi script'; ./mi_script.sh", # Comando Bash con script )

# Definir dependencias entre tareas t1 >> t2

Parámetros principales

El BashOperator tiene varios parámetros que puedes configurar:

ParámetroDescripcióntask_idIdentificador único de la tarea.bash_commandComando de Bash que se ejecutará. Puede incluir múltiples líneas.envDiccionario para configurar variables de entorno que se usarán durante la ejecución del comando.cwdCambia el directorio de trabajo donde se ejecutará el comando.execution_timeoutTiempo máximo permitido para que el comando termine antes de ser interrumpido.

Ejemplo con variables de entorno

Puedes pasar variables de entorno al comando de Bash:

t3 = BashOperator( task_id="custom_environment", bash_command="echo 'El usuario actual es: $USER'", env={"USER": "airflow_user"} # Establece la variable USER )

Logs de la ejecución

Airflow registra los logs de cada ejecución, por lo que puedes verificar los resultados de tu comando en el log del operador. Para ver los logs:

Ve al Airflow UI.
Selecciona el DAG y luego la tarea.
Haz clic en View Log.

Errores comunes

Permisos insuficientes: Asegúrate de que los scripts o comandos tengan los permisos necesarios.
Rutas incorrectas: Si usas rutas relativas, verifica que el directorio de trabajo sea el correcto.
Dependencias del sistema: Si tu comando requiere herramientas externas, verifica que estén instaladas en el entorno donde se ejecuta Airflow.

Avanzado: Plantillas Jinja

El bash_command soporta plantillas Jinja, lo que permite usar variables dinámicas como fechas de ejecución:

t4 = BashOperator( task_id="templated_command", bash_command="echo 'Fecha de ejecución: {{ ds }}'", # ds es la fecha de ejecución )

Este ejemplo imprime la fecha de ejecución del DAG (por ejemplo, 2024-11-28).

¡Con estos ejemplos deberías poder usar el BashOperator con confianza! 😊

Maria Antonia Bermudez Cardona

student•

Cecilia Gabriela Rodríguez Flores

student•

Pablo Ponte

student•

No estoy pudiendo correr ningun DAG, tengo la versión 2.3.3 corriendo dockerizada con este compose https://airflow.apache.org/docs/apache-airflow/2.3.3/docker-compose.yaml

Al parecer cuando ejecuta no puede escribir los logs, por eso falla y por eso no puedo ver el error desde la consola web.

Probe creando las carpetas dags y logs antes de correr el compose, probe dejando que las cree el airflow-init por su cuenta, mismo resultado.

Fallan todos los DAGs inlcuso los de ejemplo que vienen en airflow.

Probe con la version 2.9.0 y tengo los mismos problemas, salvo que as tasks no fallan inmediatamente pero no terminan nunca, pero en los logs del contenedor veo los mismos errores.

Estoy corriendo ubuntu 22.04 docker 26.1.1 compose 2.27.0

Gian HM

student•

[2024-03-24, 01:18:47 UTC] {taskinstance.py:1979} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: bashoperator.hello_with_bash manual__2024-03-24T01:18:43.924062+00:00 [queued]> [2024-03-24, 01:18:47 UTC] {taskinstance.py:1979} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: bashoperator.hello_with_bash manual__2024-03-24T01:18:43.924062+00:00 [queued]> [2024-03-24, 01:18:47 UTC] {taskinstance.py:2193} INFO - Starting attempt 1 of 1 [2024-03-24, 01:18:47 UTC] {taskinstance.py:2217} INFO - Executing <Task(BashOperator): hello_with_bash> on 2024-03-24 01:18:43.924062+00:00 [2024-03-24, 01:18:47 UTC] {standard_task_runner.py:60} INFO - Started process 2075 to run task [2024-03-24, 01:18:47 UTC] {standard_task_runner.py:87} INFO - Running: ['***', 'tasks', 'run', 'bashoperator', 'hello_with_bash', 'manual__2024-03-24T01:18:43.924062+00:00', '--job-id', '3', '--raw', '--subdir', 'DAGS_FOLDER/1-bashoperator.py', '--cfg-path', '/tmp/tmpzk_i2e5w'] [2024-03-24, 01:18:47 UTC] {standard_task_runner.py:88} INFO - Job 3: Subtask hello_with_bash [2024-03-24, 01:18:47 UTC] {task_command.py:423} INFO - Running <TaskInstance: bashoperator.hello_with_bash manual__2024-03-24T01:18:43.924062+00:00 [running]> on host 233392a8204a [2024-03-24, 01:18:47 UTC] {taskinstance.py:2513} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='***' AIRFLOW_CTX_DAG_ID='bashoperator' AIRFLOW_CTX_TASK_ID='hello_with_bash' AIRFLOW_CTX_EXECUTION_DATE='2024-03-24T01:18:43.924062+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-03-24T01:18:43.924062+00:00' [2024-03-24, 01:18:47 UTC] {subprocess.py:63} INFO - Tmp dir root location: /tmp [2024-03-24, 01:18:47 UTC] {subprocess.py:75} INFO - Running command: ['/usr/bin/bash', '-c', "echo 'Hello gente soy gianmarco'"] [2024-03-24, 01:18:47 UTC] {subprocess.py:86} INFO - Output: [2024-03-24, 01:18:47 UTC] {subprocess.py:93} INFO - Hello gente soy gianmarco [2024-03-24, 01:18:47 UTC] {subprocess.py:97} INFO - Command exited with return code 0 [2024-03-24, 01:18:48 UTC] {taskinstance.py:1149} INFO - Marking task as SUCCESS. dag_id=bashoperator, task_id=hello_with_bash, execution_date=20240324T011843, start_date=20240324T011847, end_date=20240324T011848 [2024-03-24, 01:18:48 UTC] {local_task_job_runner.py:234} INFO - Task exited with return code 0 [2024-03-24, 01:18:48 UTC] {taskinstance.py:3312} INFO - 0 downstream tasks scheduled from follow-on schedule check

Camilo Parra

student•

No he podido quitar los DAGs que vienen como ejemplo en el Airflow, desactivé la opción desde el docker_composer.yml AIRFLOW__CORE__LOAD_EXAMPLES: 'false' pero siguen saliendo. Ayuda!