vineyard._C.NotEnoughMemoryException

user_17588880 0
Viewed 25

I'm trying to setup airflow on vineyardd to pass dataframes from one task to another.

However, after some time with vineyard socket running, I'm facing vineyard._C.NotEnoughMemoryException which was discussed here

despite having default args as

default_args = dict(
    owner = DEFAULT_OWNER,
    depends_on_past = False,
    start_date = days_ago(1),
    on_failure_callback = partial(notify_failure_and_cleanup_xcom, users=["fip"])
)

where cleanup_xcom is

from airflow.models import XCom

@provide_session
def cleanup_xcom(context, session=None):
    logging.info(f'Clearing {XCom}')
    dag_id = context["dag"].dag_id
    XCom.clear(dag_id = dag_id)

was anyone successful with creating airflow with vineyard xcom backend?

airflow etl python
1 Answers

Resolved it with

@provide_session
def cleanup_xcom(context, session=None):
    for task in context['dag'].tasks:
        try:
            XCom.clear(task_id = task.label, dag_id = context['dag'].dag_id, execution_date = context['execution_date'])
user_175888800
Related