Python Operator

Clase 13 de 29 • Curso de Fundamentos de Apache Airflow

Jonathan Villavicencio Gonzalez

student•

This is wonderful!

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

from datetime import datetime


def print_hello():
    print('Hello people from python funtion')

with DAG(
    dag_id='python_operator',
    description='Fisrt dag using python operator',
    schedule_interval='@once',
    start_date=datetime(2022,8,1)
    ) as dag:
    
    t1 = PythonOperator(
        task_id = 'hello_with_python',
        python_callable = print_hello
    )
    ```

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

Sebastián Salas

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

Johan Fernando Astudillo

student•

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

def print_hello():
    print("HELLO PAPUS")

with DAG(dag_id="pythonoperator",
         description="Nuestro primer DAG utilizando python ope",
         schedule_interval="@once",
         start_date =datetime(2022,12,6)) as dag:


    t1 = PythonOperator(task_id="hello_with_python",
                        python_callable=print_hello)

    t1                        ```

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

Martin Macchia

student•

Leidy Johana Alarcon Moya

student•

Elitsoft Chile

student•

William Yesid Cuesta Cardona

student•

Daniel Felipe Zuluaga Cárdenas

student•

Freddy Norberto Montañez Gordillo

student•

Gerardo Mayel Fernández Alamilla

student•

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


def print_hello():
    return print('hello world with python in airflow')


with DAG(dag_id='primer_dag_python',
         schedule_interval='@once',
         start_date=datetime(2022, 7, 1)) as dag:

    t1 = PythonOperator(task_id='primer_task_python',
                        python_callable=print_hello)

Camilo Andrés Rodriguez Higuera

student•

El resultado

Julen Alvaro

student•

Hola, me funciona todo pero de una manera muy extraña:

Por un lado no encuentra nada que esté en airflow.operators y no es de extrañar, porque la carpeta "plugins" está vacía. El la carpeta gemela /opt/airflow/plugins dentro de airflow-airflow-webserver-1 tampoco he encontrado nada. Luego tarda minutos en subirse al localhost y en aparecer en la lista de DAGs (igual es que ando algo justo de RAM en local, tengo la VM de docker capada a 4GB que debería de ser suficiente). Lo curioso es que sin embargo el .log aparece:

Alguien tiene alguna explicación de por qué funciona a pesar de que parece que los archivos .py no tienen acceso a airflow.operators?

Julen Alvaro

student•

lo de que iba lento ya lo he solucionado, le faltaba RAM a Docker. He modificado el archivo .wslconfig que se encuentra en la carpeta del usuario de windows y que controla los recursos que se le asignan a Docker. Ahora lo que me va lento es el resto del Sistema Operativo, pero qué le vamos a hacer:

# Settings apply across all Linux distros running on WSL 2
[wsl2]
# Limits VM memory to use no more than 5 GB, this can be set as whole numbers using GB or MB
memory=5GB
# Sets the VM to use two virtual processors
processors=2

Lo de que el directorio plugins siga vacío y aún así el programa corra sigo sin explicármelo, todas esas referencias deben de estar en algún lugar en los contenedores de docker pero no las he podido encontrar manualmente, desde luego el directorio '/opt/airflow/plugins' de airflow-airflow-webserver-1 está vacío.

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?

Marlon Ricra

student•

Como ejecuto un script de python, que este fuera del archivo del DAG; es decir un archivo externo

Eric Bellet

teacher•

Acá el desafío es, ¿cómo conectas tu máquina con Airflow a ese fichero? Hay varias maneras, puedes hacerlo con un SSHExecuteOperator, o montas una API, rtc

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.

Python 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