Bash Operator

Clase 12 de 29 • Curso de Fundamentos de Apache Airflow

Royer Guerrero Pinilla

student•

🎮 Bash Operator

Ademas de lo mencionado en clase este se puede ejecutar cualquier comando de bash, por tanto podemos ejemplo bajar un fichero con wget o curl y luego retornar la data filtrado con grep o awk o en si ejecutar cualquier script de bash para ello solo hace falta que lo tengas dentro de la carpeta dags, (Lo unico es que script bash no pude ser ejecutado con source file.sh sino que debemos darle permisos de ejecución y llamarlo asi ./file.sh)

Gerardo Mayel Fernández Alamilla

student•

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

with DAG(dag_id='primer_dag_bash', schedule_interval='@once', start_date=datetime(2022, 7, 1)) as dag:
    t1 = BashOperator(task_id='primer_task_bash',
                      bash_command='echo "Hola mundo"')

Miguel Higuera Cardenas

student•

Eric, gracias por responder mi anterior pregunta, seguramente saldrán mas, sin embargo ahora me resulta la siguiente: como se hace para que airflow ejecute un script bash o código python que está en otro servidor; en uno de los videos comentaste que la maquina con airflow no debe si es el caso ejecutar las tareas, solo debe orquestar, sino que estas se deben ejecutar en otros servidores, ese tipo de conexión como se hace ? como haría airflow para monitorear el estatus de tareas ejecutandose en maquinas remotas ? gracias.

Eric Bellet

teacher•

Hola, una manera es usando el operador SSHExecuteOperator(). Tendrías que crear la conexión al servidor que te interesa, y ya el operador se encargaría del resto.

Jhonatan Estiven Becerra Pedrozo

student•

Tengo la misma duda

Miguel Higuera Cardenas

student•

Hola, a una tarea con bashoperator o pythonoperator se les puede asignar un bash script ya realizado o algun código en python ya hecho ? o siempre hay que escribir el codigo completo en el DAG ? gracias.

Eric Bellet

teacher•

Hola, sí puedes pasarle un fichero Python o Bash. Por ejemplo, si a un BashOperator le pasas el comando ". /file.sh" o "python file.py"

Eduard Giraldo Martínez

student•

Hola Miguel. Para la ejecución de Scripts puedes hacerlo con: > BashOperator( task_id='ejecutar_script', bash_command='python/ruta/al/script.py' )

Henrry Yánez

student•

Les comparto el código de la clase:

from airflow import DAG
from airflow.decorators import dag
from airflow.operators.bash import BashOperator

from datetime import datetime
with DAG(
dag_id="bashoperator",
description="Utilizando BASH operator",
start_date=datetime(2022, 10, 31)
) as dag:

t1 = BashOperator(task_id="hello_with_bash",
bash_command="echo 'hello bash operator'")
t1

Daniel Felipe Zuluaga Cárdenas

student•

Hola, les comparto imagen del log

Maria Antonia Bermudez Cardona

student•

Leidy Johana Alarcon Moya

student•

Este curso si que me ha traido felicidades!

Eric Bellet

teacher•

Gracias Leidy!

Cecilia Gabriela Rodríguez Flores

student•

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!

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

Luis Rodolfo Altuve Cáceres

student•

Acá esta:

Elitsoft Chile

student•

Ha cambiado un poco la interfaz, pero no mucho

Freddy Norberto Montañez Gordillo

student•

No entiendo como poner imagenes, por lo que adjunto el script creado:

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


with DAG(dag_id="bashoperator",
    description="Utilizando bash Operator",
    start_date=datetime(2022, 11, 1)    
) as dag:
    t1= BashOperator(
        task_id="hello_with_bash",
        bash_command="echo 'Hi Platzi people from bash operator'"
    )
    t1

Royer Guerrero Pinilla

student•

La arrastras desde tu sistema de archivos al comentario y esta se colocara en formato markdown, si quieres colocar una de internet usas la siguiente sintaxis

![](https://fotos.com/perrito.jpg)

https://www.markdownguide.org/getting-started/

Lucel Da Silva

student•

Sebastián Salas

student•

Shirley Tatiana Pitta Picon

student•

Camilo Andrés Rodriguez Higuera

student•

El resultado

emanuel anchique

student•

profe que extensión usa para que le aparezcan las ayudas de los métodos?, a mi no me aparecen y me toca ir directo a la pagina de la documentación

Eric Bellet

teacher•

Depende de la versión de PyCharm te reconoce las librerías del Docker. No tengo claro las versiones, pero si quieres ir por una solución segura para tener el autocompletado haz un pip install airflow en tu local

Giovedi Marmolejo

student•

No me muestra la barra para ver los log.

Eric Bellet

teacher•

Has ejecutado el proceso?

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

with DAG(dag_id='primer_dag_bash', schedule_interval='@once', start_date=datetime(2022, 7, 1)) as dag:
    t1 = BashOperator(task_id='primer_task_bash',
                      bash_command='echo "Hola mundo"')

from airflow import DAG
from airflow.decorators import dag
from airflow.operators.bash import BashOperator

from datetime import datetime
with DAG(
dag_id="bashoperator",
description="Utilizando BASH operator",
start_date=datetime(2022, 10, 31)
) as dag:

t1 = BashOperator(task_id="hello_with_bash",
bash_command="echo 'hello bash operator'")
t1

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


with DAG(dag_id="bashoperator",
    description="Utilizando bash Operator",
    start_date=datetime(2022, 11, 1)    
) as dag:
    t1= BashOperator(
        task_id="hello_with_bash",
        bash_command="echo 'Hi Platzi people from bash operator'"
    )
    t1

Bash Operator

Introducción a Apache Airflow

Introducción al curso

¿Para qué sirve airflow?

¿Por qué usar airflow?

Resumen módulo 1

Conceptos básicos

DAG

Tasks y Operators

Scheduler

Instalación y configuración

Instalación de Apache Airflow usando Docker

Posibles configuraciones

Variables y conexiones

Implementando un DAG

Implementando un DAG

Bash Operator

Python Operator

Definiendo dependencias entre tareas

Custom Operator

Orquestar y monitorizar procesos

Orquestando un DAG I

Orquestando un DAG II

Monitoring

Task Actions

Trigger Rules

Sensores

¿Qué son los sensores?

ExternalTaskSensor

FileSensor

Templates con Jinja

¿Qué son los templates con Jinja?

XComs

¿Qué son los Xcoms?

BranchPythonOperator

BranchPythonOperator

Proyecto

Definición del proyecto

Cierre del curso

Cierre del curso

Comparte tu proyecto de Platzi explora el espacio con Airflow y certifícate