8

I have tasks in pipeline to execute but those tasks are often async. I am trying to run the pipeline with Airflow but it is giving me error. "TypeError: can't pickle coroutine objects"

Since the functions are async, I thought to wrap them in "asyncio.run" but still it didn't work.

class Top:
    async def process(self, input_data):
        return [rawstr for rawstr in input_data]

class Bottom:
    async def process(self, input_data):
        return [len(x) for x in input_data]

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2019, 7, 25),
    'retries': 1,
    'provide_context': True,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG('sof_dag', default_args=default_args, schedule_interval=timedelta(days=1))

async def top_1(x, **kwargs):
    return asyncio.run(Top().process(x))

async def bottom_1(**kwargs):
    ti = kwargs['ti']
    y = ti.xcom_pull(key=None, task_ids='Router_1')
    return asyncio.run((Bottom().process(y)))

t1 = PythonOperator(
    task_id='task_top_1',
    python_callable=top_1,
    op_args=[["wow! this is great", "this is not how I thought"]],
    dag=dag)

t2 = PythonOperator(
    task_id='task_bottom_1',
    python_callable=bottom_1,
    dag=dag)

t1 >> t2

This is just a dummy scenario to give the idea how async has been used in almost every task of mine. Here is the error trace:

Traceback (most recent call last):
  File "/Users/divyanshushekhar/repos/repo_name/venv/lib/python3.7/site-packages/airflow/models/__init__.py", line 1445, in _run_raw_task
    self.xcom_push(key=XCOM_RETURN_KEY, value=result)
  File "/Users/divyanshushekhar/repos/repo_name/venv/lib/python3.7/site-packages/airflow/models/__init__.py", line 1867, in xcom_push
    execution_date=execution_date or self.execution_date)
  File "/Users/divyanshushekhar/repos/repo_name/venv/lib/python3.7/site-packages/airflow/utils/db.py", line 73, in wrapper
    return func(*args, **kwargs)
  File "/Users/divyanshushekhar/repos/repo_name/venv/lib/python3.7/site-packages/airflow/models/__init__.py", line 4460, in set
    value = pickle.dumps(value)
TypeError: can't pickle coroutine objects
[2019-08-08 18:37:09,630] {__init__.py:1603} INFO - Marking task as UP_FOR_RETRY
2
  • Were you able to figure this out? My requests are taking too long and I'd like to use aiohttp but getting the same error. Commented Nov 18, 2019 at 16:59
  • I got the same issue.@ethanenglish Have you figured this out? Commented Mar 25, 2020 at 13:26

1 Answer 1

7

You can get around it by calling a function that calls asyncio.run(), dont return a coroutine object

async def request_data(**kwargs):

    # maybe loop & create tasks then offload them other async functions

    async with session.get() as resp:

        data = await resp.text

    return data


def task_callable(**kwargs):

    cred = kwargs['params']['credentials']
    asyncio.run(request_data(session_cred=cred))


default_args = {
    owner:'yourname'
}

with DAG('my_dag', default_args=default_args, schedule = '@daily') as dag:

    task = PythonOperator(
        task_id = 'async_task',
        python_callable = task_callable,
        provide_context = True,
        params = {
            'credentials':get_cred(key_file)
    }
)
2
  • The original question is using asyncio.run() already, did you mean that the key is to make sure the callable passed in is a function that wraps any of the async functions with asyncio.run() instead of trying to make those functions also async?
    – dbailey
    Commented Mar 25, 2020 at 12:40
  • 1
    they need to remove the async from their callable function, then remove asyncio.run() as a return value. you will get the same pickling error using multiprocessing in simillar ways. Just call asyncio.run() within a callable function, dont return asyncio.run() as it is a corountine (as are all async functionalities in python) Commented Mar 25, 2020 at 14:02

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.