Execution_date not rendering after airflow upgrade

Apache Airflow version

2.2.2 (latest released)

Operating System

Debian GNU/Linux

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.3.0
apache-airflow-providers-apache-spark==2.0.1
apache-airflow-providers-cncf-kubernetes==2.1.0
apache-airflow-providers-ftp==2.0.1
apache-airflow-providers-http==2.0.1
apache-airflow-providers-imap==2.0.1
apache-airflow-providers-sqlite==2.0.1

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened

Hi,
We recently upgraded airflow from 2.1.0 to 2.2.2 (2.1.0 to 2.2.0 to 2.2.1 to 2.2.2) and DAGs aren't running as expected. All these DAGs were added before the upgrade itself and they were running fine.
We use execution_date parameter in SparkSubmitOperator which was rendering fine before the upgrade fails now returning None

"{{ (execution_date if execution_date.microsecond > 0 else dag.following_schedule(execution_date)).isoformat() }}"

DAG run fails with the error

jinja2.exceptions.UndefinedError: 'None' has no attribute 'isoformat'

Tried wiping out the database and ran as a fresh DAG but still same error

Any help would be appreciated

What you expected to happen

No response

How to reproduce

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Thanks for opening your first issue here! Be sure to follow the issue template!

Please provide a DAG for reproduction.

properties = SparkProperties.from_config("small")

def execution_date():
    return "{{ (execution_date if execution_date.microsecond > 0 else dag.following_schedule(execution_date)).isoformat() }}"

def spark_job(task_id: str,
    spark: SparkProperties,
    main_class: Optional[str],
    *args,
    py_files: Optional[str] = None,
    jars: Optional[str] = None,
    **kwargs,) -> SparkSubmitOperator:
   return SparkSubmitOperator(
        task_id=task_id,
        application=path,
        conf=spark.conf(),
        py_files=py_files,
        jars=jars,
        conn_id=spark.conn_id,
        java_class=main_class or None,
        total_executor_cores=spark.total_executor_cores,
        executor_cores=spark.executor_cores,
        executor_memory=spark.executor_memory,
        driver_memory=spark.driver_memory,
        name=task_id,
        num_executors=spark.num_executors,
        application_args=spark_args(*args, **kwargs),
        env_vars=spark.env_vars(),
    )

def test1() -> SparkSubmitOperator:
    return spark_job(
        task_id=f"test_dag",
        spark=properties,
        main_class="TestSparkJob",
        execution_date=execution_date()
    )


def test_dag(this: DAG) -> DAG:
    with this:
        (label("start") >> test1() >> label("finish"))
    return this


with DAG(
    f"test_workflow", default_args=DEFAULT_ARGS, schedule_interval="@once", max_active_runs=1,
) as dag:
    (test_dag(dag))

Any spark job would do to reproduce this issue since the DAG fails before triggering the job

Any spark job would do to reproduce this issue since the DAG fails before triggering the job

What's spark_job() doing ? This must be something internal to your organisation?

It creates a SparkSubmitOperator. I have updated the example

It creates a SparkSubmitOperator. I have updated the example

OK. Now we need to know what spark_args() does as well.

It creates a SparkSubmitOperator. I have updated the example

OK. Now we need to know what spark_args() does as well.

ah mb.. it creates application_args (List[String]) from args and kwargs arguments. Something like this

def spark_args(*args, **kwargs) -> List[str]:
    arr = [a for a in args]
    for key, value in kwargs.items():
        arr += [f"--{key}", str(value)]
    return arr

On a related note, noticed from the airflow doc that {{ execution_date }} is deprecated.
I see that {{ ts }} is equivalent to execution_date isoformat but this is a string type. Do we have an equivalent for execution_date in datetime type ?

@uranusjr

OK. Seems prety legit then. Hmm. Interesting - we converted the execution_date to be lazy-proxy object so essentially it should work in a backwards compatible way (correct @uranusjr ?).

I wonder what happen if you replace the execution_date with dag_run.logical_date - see deprecated values here https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html?highlight=macros#variables - but it should work regardless, so it would be great to track this one down.

@uranusjr

I believe I know what the problem is. yes the dag_run.logical_date should work and I know what the problem is: @uranusjr - this is the "coerce_date" in following_schedule:

    def following_schedule(self, dttm):
        """
        Calculates the following schedule for this dag in UTC.

        :param dttm: utc datetime
        :return: utc datetime
        """
        warnings.warn(
            "`DAG.following_schedule()` is deprecated. Use `DAG.next_dagrun_info(restricted=False)` instead.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        data_interval = self.infer_automated_data_interval(timezone.coerce_datetime(dttm))
        next_info = self.next_dagrun_info(data_interval, restricted=False)
        if next_info is None:
            return None
        return next_info.data_interval.start

Here is coerce_date:


def coerce_datetime(v: Union[None, dt.datetime, DateTime]) -> Optional[DateTime]:
    """Convert whatever is passed in to an timezone-aware ``pendulum.DateTime``."""
    if v is None:
        return None
    if v.tzinfo is None:
        v = make_aware(v)
    if isinstance(v, DateTime):
        return v
    return pendulum.instance(v)

I can see why the error happens, but not why the error is wrong. A @once schedule does not have a following schedule (because it runs only once?), so dag.following_schedule(execution_date) is correct to return None. This also should be the case if you use dag_run.logical_date.

Ah. I see the @once indeed you arre right.

If schedule_interval is a valid case, is it coerce_date issue ?
Could you confirm if dag_run.logical_date would work for this ?

When the DAGs are manually triggered(after a couple of runs), it renders the date. But in scheduled runs, it always fails returning None

Could you confirm if dag_run.logical_date would work for this ?

Can you try it pleaase? I cannot "confirm it" but if it works that will give us a clue to look further

Could you confirm if dag_run.logical_date would work for this ?

Can you try it pleaase? I cannot "confirm it" but if it works that will give us a clue to look further

dag_run.logical_date didn't work either

Narrowing down further, dag.following_schedule(execution_date) returns null for scheduled sub dags only. Main dags return value

Oh I know what’s going on now. We’ve been looking at entirely wrong places.

The change that actually broke the logic is #16141. Previously, a manually-triggered run always has its execution date set to timezone.now(), which has an approximately 99.9999% chance to return a non-zero millisecond, so

(execution_date if execution_date.microsecond > 0 else dag.following_schedule(execution_date)).isoformat()

is (almost) always evaluated to

execution_date.isoformat()

which works.

But the calendar widget introduced in #16141 does not employ sub-second accurary, so the milliseconds become zero, and all manually-triggered runs now evaluate the expression to

dag.following_schedule(execution_date).isoformat()

instead, which always fails because a manually triggered run never has a following schedule (both prior and after AIP-39).

Ultimately, using the millisecond field to distinguish between scheduled and manual runs is a poor implementation in the first place. There’s still a 0.0001% chance the detection would fail, even if Airflow never broke that. It is both more semantically correct and reliable to use dag_run.run_type instead.

I think the best we can do here is to add an entry to UPDATING.md to describe this behavioural change.

One correction. (execution_date if execution_date.microsecond > 0 else dag.following_schedule(execution_date)).isoformat()
fails for scheduled runs as well. dag.following_schedule(execution_date) returns null for scheduled sub dags

Even if we use dag_run.run_type to figure out the run type, what is the right way to get execution_date for scheduled runs ?

execution_date will still work until 3.0. logical_date is the new variable name (the value is the same). But by the fact you’re using following_schedule, I suspect data_interval_end actually provides more suitable semantics for you use case.

But the calendar widget introduced in #16141 does not employ sub-second accurary, so the milliseconds become zero, and all manually-triggered runs now evaluate the expression to dag.following_schedule(execution_date).isoformat()
instead, which always fails because a manually triggered run never has a following schedule (both prior and after AIP-39).

also this reasoning doesn't hold true for this issue since the error is seen for scheduled sub dags

@uranusjr

I think the best we can do here is to add an entry to UPDATING.md to describe this behavioural change.

@uranusjr was this added? seems like there is no actual task on this issue other than clarification?

No I don’t think so. I’ll do that.

also this reasoning doesn't hold true for this issue since the error is seen for scheduled sub dags

Sorry I missed this. A sub-DAG’s execution date is inherited from its parent DAG, so the behaviour change in the parent DAG creeps down to its sub-DAGs.

Navigation Menu

Repository navigation

Execution_date not rendering after airflow upgrade #19618

Apache Airflow version

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

What happened

What you expected to happen

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

11 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Execution_date not rendering after airflow upgrade #19618

Description

Apache Airflow version

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

What happened

What you expected to happen

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Activity

boring-cyborg commented on Nov 16, 2021

uranusjr commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

potiuk commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

potiuk commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

potiuk commented on Nov 16, 2021

potiuk commented on Nov 16, 2021

uranusjr commented on Nov 16, 2021

potiuk commented on Nov 16, 2021

11 remaining items

AnithaGG commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

potiuk commented on Nov 16, 2021

AnithaGG commented on Nov 16, 2021

AnithaGG commented on Nov 17, 2021

uranusjr commented on Nov 17, 2021

AnithaGG commented on Nov 17, 2021

uranusjr commented on Nov 17, 2021

AnithaGG commented on Nov 17, 2021

eladkal commented on Jun 13, 2022

uranusjr commented on Jun 13, 2022

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions