Sets in parameters cause "TypeError: Object of type set is not JSON serializable"

[MatrixManAtYrService]

Apache Airflow version

2.2.1 (latest released)

Operating System

Debian Buster (docker)

Versions of Apache Airflow Providers

n/a

Deployment

Astronomer

Deployment details

Using astro dev start

No problems with:

FROM quay.io/astronomer/ap-airflow:2.1.4-buster-onbuild

Problems occur with:

FROM quay.io/astronomer/ap-airflow:2.2.0-buster-onbuild

The DAG:

@task(params={"a": {1, 2, 3}, "b": {3, 4, 5}})
def intersect(intersection):
    print(intersection)

@dag(
    schedule_interval="@daily",
    start_date=days_ago(1),
    default_args={"owner": "airflow"},
)
def set_params():
    intersect("{{ params.a.intersection(params.b).pop() }}")

the_dag = set_params()

What happened

The DAG parses and runs fine on 2.1.4, but when I upgrade to 2.2.0 I see an error at the top of the DAGs list:

The dag is still listed (from successful runs in 2.1.4) and when I click it I see a new error:

Python version: 3.9.7
Airflow version: 2.2.0+astro.2
Node: 10b491f393be
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python3.9/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/auth.py", line 51, in decorated
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/decorators.py", line 109, in view_func
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/decorators.py", line 72, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/views.py", line 2259, in tree
    dag = current_app.dag_bag.get_dag(dag_id)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py", line 186, in get_dag
    self._add_dag_from_db(dag_id=dag_id, session=session)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py", line 261, in _add_dag_from_db
    dag = row.dag
  File "/usr/local/lib/python3.9/site-packages/airflow/models/serialized_dag.py", line 180, in dag
    dag = SerializedDAG.from_dict(self.data)  # type: Any
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 916, in from_dict
    return cls.deserialize_dag(serialized_obj['dag'])
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 830, in deserialize_dag
    v = {task["task_id"]: SerializedBaseOperator.deserialize_operator(task) for task in v}
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 830, in <dictcomp>
    v = {task["task_id"]: SerializedBaseOperator.deserialize_operator(task) for task in v}
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 587, in deserialize_operator
    v = cls._deserialize_operator_params(v)
  File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 729, in _deserialize_operator_params
    param_class = import_string(v['__class'])
KeyError: '__class'

What you expected to happen

The dag runs the same as it did in 2.1.4

How to reproduce

Run the above DAG in 2.2.0

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

[ephraimbuddy]

The dag above doesn't appear on the UI and doesn't show import errors, probably needs modification. However, I saw the error from:

from datetime import datetime
from airflow.models import DAG


dag = DAG("paramtest",
          start_date=datetime(2021, 1, 1),
          schedule_interval="@daily")


@dag.task(params={"a": {1, 2, 3}, "b": {3, 4, 5}})
def intersect(intersection):
    print(intersection)

intersect("{{ params.a.intersection(params.b).pop() }}")

[MatrixManAtYrService]

If my database contains dagruns from when this used to work (1.10.15 for instance) then I can still run the dag despite the error above. When I do, I get this error:

Python version: 3.9.7
Airflow version: 2.2.1.dev814+astro.1
Node: cfe83c6f54fc
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python3.9/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.9/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/auth.py", line 51, in decorated
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/decorators.py", line 72, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/www/views.py", line 1721, in trigger
    if unpause and dag.is_paused:
AttributeError: 'NoneType' object has no attribute 'is_paused'

[MatrixManAtYrService]

Using these DAGs: https://gist.github.com/MatrixManAtYrService/c56bb8b94e40064ece1a6662ce585670

I saw similar behavior, except instead the error was:

TypeError: Object of type NoValueSentinel is not JSON serializable

(FYI: @jedcunningham)

[dstandish]

Taking a look at this. But first wanted to establish a workaround. And this seems to work:

from datetime import datetime
from typing import Any

from airflow.models import DAG, BaseOperator


dag = DAG("paramtest",
          start_date=datetime(2021, 1, 1),
          schedule_interval="@daily", catchup=False, user_defined_macros={'set': lambda x: set(x)})


@dag.task(params={"a": [1, 2, 3], "b": [3, 4, 5]})
def intersect(intersection):
    print(intersection)


intersect("{{ set(params.a).intersection(params.b).pop() }}")

Notice that you need to add set as a macro, because builtins are not available in jinja. But this allows you to use those same set operations in a template.

[dstandish]

OK so, I found the source of the the problem but need to look into it more to see what we need to do.

Serialization blows up here because it's not json encodable:

https://github.com/apache/airflow/blob/main/airflow/models/serialized_dag.py#L97

And the reason why is the set is not serialized but left as a set.

And the reason why this happens might be here:

https://github.com/apache/airflow/blob/main/airflow/serialization/serialized_objects.py#L420

It doesn't use cls._serialize. We could change it to do so, but then we wouldn't be able to deserialize it back to set. I am not sure why we have a separate process for params. Maybe we can somehow unify them.

In any case I have to look into this more to see how we might be able to ensure that params are properly serialized.

[ajbosco]

@jedcunningham The issue in #19267 was found in 2.2.2rc1.

[dstandish]

dang, let me try to repro

[dstandish]

@ajbosco can you provide a minimal sample dag?