Python Operator

Clase 13 de 29 • Curso de Fundamentos de Apache Airflow

Clase anteriorSiguiente clase

Camilo Andrés Rodriguez Higuera

student•

El resultado

Jose Daniel Velasquez H

company_admin•

rogelio cortez

student•

from airflow import DAGfrom datetime import datetimefrom airflow.operators.python import PythonOperator

#If you need access to Airflow context variables (e.g., ds, run_id), you must include **kwargs.def hello(country): #**kwaargs): print(f"Hello {country}") with DAG (dag_id="pythonOperator", description="Python Operator DAG", start_date=datetime(2025, 3, 27), schedule="@once") as dag: t1 = PythonOperator(task_id='taskPython', python_callable=hello,op_args=["Mexico"]) t2 = PythonOperator(task_id='taskPython2', python_callable=hello,op_kwargs={'countrty':"France"}) # op_args = you can pass a list of arguments to the function. if you have more than one argument you must pass the arguments in order in a list# op_kwargs = you can pass a dictionary where the keys are the name of the argumnet ant the values are the values you want to pass. if you have more than one argument you can pass them in any order.

Cristian Decaroli

student•

Cristian Decaroli

student•

Python Operator successfull!!

Gino G. Viloria

student•

Daniel Olave

student•

Mario Alexander Vargas Celis

student•

Néstor Miguel González Flores

student•

¿Cómo ejecuto SQLAlchemy 2 en los DAGs usando PythonOperatos? dado que Airflow usa la versión 1.4. pasa que tengo toda una estructura de código que está en un árbol de directorios por orden para hacer un ETL, y si usara el PythonEnvOperator no me toda los demás archivos.

Cecilia Gabriela Rodríguez Flores

student•

Álvaro José Polo Franco

student•

from airflow import DAGfrom airflow.operators.python import PythonOperatorfrom datetime import datetime def print_hello(): print("Hola de pruebas.") with DAG(dag_id="pythonoperator", description="Nuestro primer DAG utilizando python operator", schedule_interval="@once", start_date=datetime(2022, 8, 1)) as dag: t1 = PythonOperator( task_id="hello_with_python", python_callable=print_hello ) t1

Álvaro José Polo Franco

student•

Jesús Zelaya Contreras

student•

Martin Macchia

student•

Leidy Johana Alarcon Moya

student•

Andres Troaños

student•

Ejecute ambos dags y ninguno de los dos me da la opción del pequeño cuadro verder y ambos aparecen como success, tampoco puedo ver lo logs

Alguien sabe que debo hacer o que sucede?

Eric Bellet

teacher•

Me muestras el código?

Luis Rodolfo Altuve Cáceres

student•

Ahora con Python Operator:

Eric Bellet

teacher•

Buen trabajo Luis

Elias Dudamel

student•

Me gusta la parte de que se se sigan podiendo reutilizar funciones, dejo ejemplo

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def print_hello(country, **kwargs):
    print(f'I am processing this {country}')


with DAG(dag_id='pythonoperator'
        , description='Utilizando python operator'
        , start_date=datetime(2023, 6, 7)
        , schedule_interval='@once') as dag:


    t1 = PythonOperator(task_id='process_ar'
                        , python_callable=print_hello
                        , op_kwargs={'country':'AR'})

    t2 = PythonOperator(task_id='process_uy'
                        , python_callable=print_hello
                        , op_kwargs={'country':'UY'})

    t1 >> t2

Elitsoft Chile

student•

Héctor Javier Herrera Espínola

student•

Me sale este error: Probé instalar de mil maneras, en todas igual:

Eric Bellet

teacher•

No me sale el error en tu pregunta