Operatori in Apache Airflow: Bash, Python e SQL

Gli Operatori di Apache Airflow sono il cuore della piattaforma e permettono di eseguire vari tipi di task all’interno dei DAG. In questo articolo, esploreremo i tre operatori più utilizzati: BashOperator, PythonOperator e SqlOperator, fornendo esempi pratici di utilizzo per automatizzare flussi di lavoro.

Cosa sono gli operatori in Apache Airflow?

Un operatore in Airflow rappresenta un singolo task all’interno di un DAG. Ogni operatore esegue una specifica operazione, che può variare da un semplice script bash, a una funzione Python o una query SQL. Airflow fornisce una vasta gamma di operatori per soddisfare le diverse esigenze di automazione e orchestrazione dei processi.

Utilizzare il BashOperator

Il BashOperator viene utilizzato per eseguire comandi o script bash all’interno dei DAG. È utile quando si ha bisogno di eseguire operazioni a livello di sistema, come lanciare script di shell, controllare file di log o avviare servizi. Ecco un esempio di come utilizzarlo:

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

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 11, 1),
}

dag = DAG('bash_operator_example', default_args=default_args, schedule_interval='@daily')

bash_task = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag
)

In questo esempio, il task bash_task esegue il comando date, che stampa la data e l’ora corrente. Il task verrà eseguito quotidianamente grazie al parametro schedule_interval='@daily'.

Utilizzare il PythonOperator

Il PythonOperator è uno degli operatori più potenti, poiché consente di eseguire funzioni Python all’interno dei DAG. È ideale per automatizzare processi che richiedono logica personalizzata o elaborazione dei dati. Ecco un esempio:

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

def hello_world():
    print("Ciao dal PythonOperator!")

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 11, 1),
}

dag = DAG('python_operator_example', default_args=default_args, schedule_interval='@daily')

python_task = PythonOperator(
    task_id='say_hello',
    python_callable=hello_world,
    dag=dag
)

In questo esempio, il task python_task esegue la funzione hello_world, che stampa un messaggio. È possibile sostituire hello_world con qualsiasi funzione Python personalizzata per automatizzare le operazioni.

Utilizzare il SqlOperator

Il SqlOperator è progettato per eseguire query SQL su un database. Può essere utilizzato per inserire, aggiornare o selezionare dati da una tabella. Ecco un esempio di come eseguire una query SQL con questo operatore:

from airflow import DAG
from airflow.operators.sql import SqlOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 11, 1),
}

dag = DAG('sql_operator_example', default_args=default_args, schedule_interval='@daily')

sql_task = SqlOperator(
    task_id='create_table',
    sql='CREATE TABLE IF NOT EXISTS example_table (id INT, name VARCHAR(50));',
    conn_id='my_sql_connection',
    dag=dag
)

In questo esempio, il task sql_task esegue una query per creare una tabella in un database MySQL (definito dalla connessione conn_id='my_sql_connection').

Conclusione

Gli Operatori di Apache Airflow ti permettono di automatizzare task complessi con facilità. Che tu debba eseguire comandi di sistema, script Python o query SQL, Airflow ha l’operatore giusto per ogni esigenza. Ora che conosci i tre operatori più comuni, puoi iniziare a costruire flussi di lavoro più complessi e personalizzati.

Fonte: Documentazione ufficiale di Apache Airflow

Per altri tutorial, visita la sezione: Tutti gli articoli