Modifica del fuso orario di un DAG su Amazon MWAA - Flussi di lavoro gestiti da Amazon per Apache Airflow
VersionePrerequisitiAutorizzazioniCrea un plug-in per modificare il fuso orario nei registri di AirflowCreazione di una plugins.zipEsempio di codiceFasi successive

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Modifica del fuso orario di un DAG su Amazon MWAA

Apache Airflow pianifica il grafo aciclico diretto (DAG) in UTC+0 per impostazione predefinita. I passaggi seguenti illustrano come modificare il fuso orario in cui Amazon MWAA esegue il tuo con Pendulum. DAGs Facoltativamente, questo argomento dimostra come creare un plug-in personalizzato per modificare il fuso orario per i log Apache Airflow del tuo ambiente.

Versione

È possibile utilizzare l'esempio di codice in questa pagina con Apache Airflow v2 in Python 3.10 e Apache Airflow v3in Python 3.11.

Prerequisiti

Per utilizzare il codice di esempio in questa pagina, avrai bisogno di quanto segue:

Autorizzazioni

Non sono necessarie autorizzazioni aggiuntive per utilizzare l'esempio di codice in questa pagina.

Crea un plug-in per modificare il fuso orario nei registri di Airflow

Apache Airflow eseguirà i file Python nella directory all'avvio. plugins Con il seguente plugin, puoi sovrascrivere il fuso orario dell'esecutore, che modifica il fuso orario in cui Apache Airflow scrive i log.

  1. Crea una directory con il nome del tuo plugin personalizzato e accedi plugins alla directory. Ad esempio:

    $ mkdir plugins
$ cd plugins

  2. Copiate il contenuto del seguente esempio di codice e salvatelo localmente come dag-timezone-plugin.py nella plugins cartella.

    import time
import os

os.environ['TZ'] = 'America/Los_Angeles'
time.tzset()

  3. Nella plugins directory, crea un file Python vuoto denominato. __init__.py La tua plugins directory dovrebbe essere simile alla seguente:

    plugins/
 |-- __init__.py
 |-- dag-timezone-plugin.py

Creazione di una plugins.zip

I passaggi seguenti spiegano come creareplugins.zip. Il contenuto di questo esempio può essere combinato con altri plugin e file binari in un unico plugins.zip file.

  1. Nel prompt dei comandi, accedete alla plugins directory del passaggio precedente. Ad esempio:

    cd plugins

  2. Comprimi il contenuto all'interno della plugins cartella.

    zip -r ../plugins.zip ./

  3. Caricalo plugins.zip nel tuo bucket S3

    aws s3 cp plugins.zip s3://your-mwaa-bucket/

Esempio di codice

Per cambiare il fuso orario predefinito (UTC+0) in cui viene eseguito il DAG, useremo una libreria chiamata Pendulum, una libreria Python per lavorare con datetime che riconosce il fuso orario.

  1. Nel prompt dei comandi, accedi alla directory in cui sono archiviati i tuoi. DAGs Ad esempio:

    cd dags

  2. Copiate il contenuto del seguente esempio e salvatelo con nome. tz-aware-dag.py

    from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
# Import the Pendulum library.
import pendulum

# Instantiate Pendulum and set your timezone.
local_tz = pendulum.timezone("America/Los_Angeles")

with DAG(
    dag_id = "tz_test",
    schedule_interval="0 12 * * *",
    catchup=False,
    start_date=datetime(2022, 1, 1, tzinfo=local_tz)
) as dag:
    bash_operator_task = BashOperator(
        task_id="tz_aware_task",
        dag=dag,
        bash_command="date"
    )

  3. Esegui il AWS CLI comando seguente per copiare il DAG nel bucket del tuo ambiente, quindi attiva il DAG utilizzando l'interfaccia utente di Apache Airflow. 

    aws s3 cp your-dag.py s3://your-environment-bucket/dags/

  4. In caso di successo, otterrete un risultato simile al seguente nei log delle attività per il DAG: tz_aware_task tz_test

    [2022-08-01, 12:00:00 PDT] {{subprocess.py:74}} INFO - Running command: ['bash', '-c', 'date']
[2022-08-01, 12:00:00 PDT] {{subprocess.py:85}} INFO - Output:
[2022-08-01, 12:00:00 PDT] {{subprocess.py:89}} INFO - Mon Aug  1 12:00:00 PDT 2022
[2022-08-01, 12:00:00 PDT] {{subprocess.py:93}} INFO - Command exited with return code 0
[2022-08-01, 12:00:00 PDT] {{taskinstance.py:1280}} INFO - Marking task as SUCCESS. dag_id=tz_test, task_id=tz_aware_task, execution_date=20220801T190033, start_date=20220801T190035, end_date=20220801T190035
[2022-08-01, 12:00:00 PDT] {{local_task_job.py:154}} INFO - Task exited with return code 0
[2022-08-01, 12:00:00 PDT] {{local_task_job.py:264}} INFO - 0 downstream tasks scheduled from follow-on schedule check

Fasi successive