AWS MWAA环境下Airflow DAG通过SSHOperator传递指定时间格式遇阻求助

问题原因 #

你用datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%f%z')失败的核心原因是：utcnow()返回的是无时区信息的naive datetime对象，而%z格式符要求datetime必须包含时区属性，因此无法正常解析，导致输出异常或报错。

解决方案 #

不需要额外安装Pendulum，用Python标准库的datetime.timezone即可构造带UTC时区的datetime对象，以下是两种可行实现方式：

方式1：通过PythonOperator生成时间再传递

import datetime
from airflow.operators.python import PythonOperator
from airflow.providers.ssh.operators.ssh import SSHOperator

def generate_formatted_utc_time():
    # 创建带UTC时区的datetime对象
    aware_utc_now = datetime.datetime.now(datetime.timezone.utc)
    # 按要求格式输出，%z会自动生成+0000（UTC时区标识）
    return aware_utc_now.strftime('%Y-%m-%dT%H:%M:%S.%f%z')

# 先执行Python任务生成时间
get_time_task = PythonOperator(
    task_id='generate_utc_time',
    python_callable=generate_formatted_utc_time,
    dag=dag
)

# 通过SSHOperator调用脚本并传递时间参数
run_script_task = SSHOperator(
    task_id='execute_shell_script',
    ssh_conn_id='your_ssh_connection_id',  # 替换为你的SSH连接ID
    command='/path/to/your/target_script.sh {{ ti.xcom_pull(task_ids="generate_utc_time") }}',
    dag=dag
)

# 设置任务依赖
get_time_task >> run_script_task

方式2：直接在SSHOperator中用Airflow模板变量（更简洁）

利用Airflow内置的execution_date模板变量（自带时区信息），直接转换为UTC并格式化：

from airflow.providers.ssh.operators.ssh import SSHOperator

run_script_task = SSHOperator(
    task_id='execute_shell_script',
    ssh_conn_id='your_ssh_connection_id',
    command='''
        # 用Airflow模板变量生成指定格式的UTC时间
        FORMATTED_UTC="{{ execution_date.in_timezone('UTC').strftime('%Y-%m-%dT%H:%M:%S.%f%z') }}"
        # 传递给Shell脚本
        /path/to/your/target_script.sh $FORMATTED_UTC
    ''',
    dag=dag
)

验证方式 #

你可以先在DAG中添加一个PythonOperator测试格式化是否正常：

import datetime
from airflow.operators.python import PythonOperator

def test_time_format():
    aware_utc_now = datetime.datetime.now(datetime.timezone.utc)
    formatted = aware_utc_now.strftime('%Y-%m-%dT%H:%M:%S.%f%z')
    print(f"Formatted UTC time: {formatted}")
    return formatted

test_task = PythonOperator(
    task_id='test_time_formatting',
    python_callable=test_time_format,
    dag=dag
)

内容的提问来源于stack exchange，提问作者PythonDeveloper