Bash Operator
Contenido del curso
Conceptos básicos
Instalación y configuración
Implementando un DAG
Orquestar y monitorizar procesos
Sensores
Templates con Jinja
XComs
BranchPythonOperator
Proyecto
Cierre del curso
Bash Operator
Contenido del curso
Bash Operator
Royer Guerrero Pinilla
Estudiante
Gerardo Mayel Fernández Alamilla
Estudiante
Miguel Higuera Cardenas
Estudiante
Eric Bellet
Profesor
Jhonatan Estiven Becerra Pedrozo
EstudianteMiguel Higuera Cardenas
EstudianteEric Bellet
Profesor
Eduard Giraldo Martínez
Estudiante
Henrry Yánez
Estudiante
Daniel Felipe Zuluaga Cárdenas
Estudiante
Maria Antonia Bermudez Cardona
Estudiante
Leidy Johana Alarcon Moya
EstudianteEric Bellet
Profesor
Cecilia Gabriela Rodríguez Flores
EstudianteCamilo Parra
Estudiante
Gian HM
Estudiante
Luis Rodolfo Altuve Cáceres
EstudianteElitsoft Chile
Estudiante
Freddy Norberto Montañez Gordillo
EstudianteRoyer Guerrero Pinilla
Estudiante
Lucel Da Silva
Estudiante
Sebastián Salas
Estudiante
Javier Ladino
EstudianteGeisemberg Marcos Zamora Yuave
Estudiante
Shirley Tatiana Pitta Picon
Estudiante
Camilo Andrés Rodriguez Higuera
Estudiante🎮 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)
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"')
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.
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.
Tengo la misma duda
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.
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"
Hola Miguel. Para la ejecución de Scripts puedes hacerlo con: > BashOperator( task_id='ejecutar_script', bash_command='python/ruta/al/script.py' )
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
Hola, les comparto imagen del log
Este curso si que me ha traido felicidades!
Gracias Leidy!
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!
[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
Acá esta:
Ha cambiado un poco la interfaz, pero no mucho
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
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
El resultado
