Non-JSON-serializable params deprecated in Airflow 2.3.0. How should I pass non-Json-serializable objects now?

user_13183351 0
Viewed 216

I am currently using the params kwarg in my DAG objects to pass extra configurations to my tasks, which are PythonDecoratedOperators.

These configurations (stored in python dictionaries) include datetime objects or even lambda functions that I'm able to handle from a configuration file. This allows me to easily change some important aspects without touching the rest of my code.

This practice has been deprecated in Airflow 2.3.0 and will be removed in future versions.

My question is: how should I proceed in the future? Is there a better way to handle this?

A very simple example in a single file would be:

# Packages
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from airflow.operators.python import get_current_context

import logging

# Here is my configuration dict.
config = {
    'value': 5,
    'operation': lambda x: x**2
}


default_args = {
    'start_date': days_ago(1)
}

@dag(schedule_interval=None, default_args=default_args, catchup=False, params=config) # I pass my config dict using the params kwarg.
def dag_example():

    @task
    def extract_message(**kwargs) -> int:
        context = get_current_context()
        config = context['params']

        value = config['value']

        logging.info('The value is:' + str(value))

        return value

    @task
    def process_data(value: int) -> int:
        context = get_current_context()
        config = context['params']

        operation = config['operation']

        value = operation(value)

        return value

    @task
    def store_data(value: int) -> None:
        with open('/opt/airflow/data/try1.text', 'a+') as file:
            print(value, file=file)


    store_data(process_data(extract_message()))

dag_example = dag_example()

if __name__ == '__main__':
  from airflow.utils.state import State
  dag_example.clear(dag_run_state=State.NONE)
  dag_example.run()

With this, I would obviously receive the following warning:

[2022-05-04, 13:09:26 CEST] {warnings.py:109} WARNING - /home/---/.local/lib/python3.8/site-packages/---/models/param.py:59: DeprecationWarning: The use of non-json-serializable params is deprecated and will be removed in a future release

airflow-2.x airflow python
1 Answers

My naive solution would be to pass config as a param in your python functions.

I would do something like this (note I renamed value because it was being overridden):

  @task
    def process_data(key, config) -> int:
        operation = config['operation']
        value = operation(key)
        return value

process_data(key=extract_message(), config=config)
user_185043390
Related