Как запустить конкретную дату в прошлом, которая находится вне временной шкалы? - PullRequest
Новая
       19

Как запустить конкретную дату в прошлом, которая находится вне временной шкалы?

0 голосов
/ 08 июля 2019

Я пытаюсь запустить определенную дату в прошлом, которая находится за пределами временной шкалы истории запусков.

Например, прикрепленный рисунок: тест воздушного потока {1} показывает прогоны каждый день, с 3 по 8 июля. Однако мы решили, что нам нужно пробежать 29 июня.

Я пытался изменить некоторые параметры, но это не помогло, оно продолжало выполняться с сегодняшнего дня 8 июля. Рисунок: Проверка воздушного потока {2}

Я попытался заменить на следующие параметры:

'start_date': datetime.strptime ("2019-06-29", "% Y-% m-% d"),

schedule_interval = "@ один раз"

Вот код: 

    from datetime import datetime, timedelta, date as dt

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

    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args = {
        'owner': 'airflow',
        'depends_on_past': False,
        'start_date': datetime.strptime("2019-06-29", "%Y-%m-%d"),#airflow.utils.dates.days_ago(5),
        'email': ['airflow@example.com'],
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'dag': dag,
        # 'sla': timedelta(hours=2),
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function,
        # 'on_success_callback': some_other_function,
        # 'on_retry_callback': another_function,
        # 'sla_miss_callback': yet_another_function,
        # 'trigger_rule': 'all_success'
    }

    dag = DAG(
        'tutorial',
        default_args=default_args,
        description='A simple tutorial DAG',
        schedule_interval="@once"#timedelta(days=1),
    )

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    t1 = BashOperator(
        task_id='print_date',
        bash_command='date',
        dag=dag,
    )

    t1.doc_md = """\


    dag.doc_md = __doc__

    t2 = BashOperator(
        task_id='sleep',
        depends_on_past=False,
        bash_command='sleep 5',
        dag=dag,
    )

    templated_command = """
            {% for i in range(5) %}
                echo "{{ ds }}"
                echo "{{ macros.ds_add(ds, 7)}}"
                echo "{{ params.my_param }}"
            {% endfor %}
    "        ""

    t3 = BashOperator(
        task_id='templated',
        depends_on_past=False,
        bash_command=templated_command,
        params={'my_param': 'Parameter I passed in'},
        dag=dag,
    )

    t1 >> [t2, t3]"

Ожидается: эта конкретная дата запускается 29 июня, а также сохраняется история запусков.

1 Ответ

1 голос
/ 08 июля 2019

Я могу сделать это с помощью CLI.

обратная заправка воздушного потока.

учебное пособие по обратной заправке воздушного потока -s 2019-06-29 -e 2019-06-29

ссылка: https://airflow.apache.org/cli.html#backfill

...