Question about DAG pickling

[sakaia]

In my environment, PythonOperator works with airflow backfill -x. but it fails without -x (i.e. airflow backfill).

In my understanding, it generates the pickle object in somewhere.
And it reads on airflow-worker.

For my case, I cannot found pickle nothing on dags volume.
To work appropriate pickle, where should I investigate?

[XD-DENG]

Hi @sakaia , it will be helpful if you can share the detailed error log/traceback of the failure you mentioned

[sakaia]

Thank you for commenting.
I failed following python code with KubernetesExeccutor. In this environment Bash Operator wirks fine.

# Airflow python test

from airflow.utils.dates import days_ago
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

af_run_dag_default_args = {
    'owner': 'sakaia'
}

## Define DAG details
af_run_dag = DAG(
    dag_id='pythontest',
    default_args=af_run_dag_default_args,
    schedule_interval=None,
    start_date=days_ago(2),
    tags=['training']
)

def pythonTest(**kwargs) -> str :
    print("pythonTest")
    return "pythonTest"

# Define DAG steps/workflow
with af_run_dag as dag :
    python_test = PythonOperator(
         task_id='python_test',
         python_callable=pythonTest,
         op_kwargs={
         },
         dag=dag
    )

Logs are follows (kubectl logs)

Loading pickle id  2
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 37, in <module>
    args.func(args)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/cli.py", line 81, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/bin/cli.py", line 618, in run
    dag_pickle = session.query(DagPickle).filter(DagPickle.id == args.pickle).first()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 3429, in first
    ret = list(self[0:1])
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 3203, in __getitem__
    return list(res)
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/loading.py", line 100, in instances
    cursor.close()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
    compat.raise_(
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/loading.py", line 80, in instances
    rows = [proc(row) for row in fetch]
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/loading.py", line 80, in <listcomp>
    rows = [proc(row) for row in fetch]
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/loading.py", line 579, in _instance
    _populate_full(
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/loading.py", line 725, in _populate_full
    dict_[key] = getter(row)
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process
    return loads(value)
  File "/home/airflow/.local/lib/python3.8/site-packages/dill/_dill.py", line 283, in loads
    return load(file, ignore, **kwds)
  File "/home/airflow/.local/lib/python3.8/site-packages/dill/_dill.py", line 278, in load
    return Unpickler(file, ignore=ignore, **kwds).load()
  File "/home/airflow/.local/lib/python3.8/site-packages/dill/_dill.py", line 481, in load
    obj = StockUnpickler.load(self)
  File "/home/airflow/.local/lib/python3.8/site-packages/dill/_dill.py", line 471, in find_class
    return StockUnpickler.find_class(self, module, name)
ModuleNotFoundError: No module named 'unusual_prefix_11b7d82e693987284b24d76e40f5bf095f51f021_pythontest'

[turbaszek]

@sakaia if you are using Airflow 2.0 your problem can be related to #14092

[sakaia]

Thank you for your suggestion. I am currently using 1.10.12 and 1.10.14