ValueError: unsupported pickle protocol: 5

[kn6405]

Apache Airflow version

2.1.2

Operating System

Official docker image

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

Originally had airflow 2.0.2 installed via 3rd party helm chart (running external postgres db). I uninstalled it and installed airflow 2.1.2 using the official chart. Pods are alll started up except the schedular pod is crashing with "ValueError: unsupported pickle protocol: 5"

What happened

I get a "ValueError: unsupported pickle protocol: 5". I went into the database and did some cleanup but still getting error. Not sure if there are other tables/colums that also need to be cleaned up.

Cleanup Done:

UPDATE public.dag_run SET conf= NULL;
DELETE FROM public.dag_pickle;

Logs:

---

____ |( )_______ / /________ __
____ /| |_ /__ / / __ / __ _ | /| / /
___ ___ | / _ / _ / _ / / // / |/ |/ /
// |// // // // _/____/|__/
[2021-08-31 16:20:29,139] {scheduler_job.py:1266} INFO - Starting the scheduler
[2021-08-31 16:20:29,139] {scheduler_job.py:1271} INFO - Processing each file at most -1 times
[2021-08-31 16:20:29,523] {dag_processing.py:254} INFO - Launched DagFileProcessorManager with pid: 19
[2021-08-31 16:20:29,528] {scheduler_job.py:1835} INFO - Resetting orphaned tasks for active dag runs
[2021-08-31 16:20:29,539] {settings.py:51} INFO - Configured default timezone Timezone('UTC')
[2021-08-31 16:20:29,628] {scheduler_job.py:1315} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1299, in _execute
self._run_scheduler_loop()
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1392, in _run_scheduler_loop
num_queued_tis = self._do_scheduling(session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1531, in _do_scheduling
self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1759, in _schedule_dag_run
self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1785, in verify_integrity_if_dag_changed
dag_run.verify_integrity(session=session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 638, in verify_integrity
tis = self.get_task_instances(session=session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 328, in get_task_instances
return tis.all()
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3373, in all
return list(self)
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 100, in instances
cursor.close()
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in exit
with_traceback=exc_tb,
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise
raise exception
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in instances
rows = [proc(row) for row in fetch]
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in
rows = [proc(row) for row in fetch]
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 588, in _instance
populators,
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 725, in populate_full
dict[key] = getter(row)
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process
return loads(value)
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 275, in loads
return load(file, ignore, **kwds)
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 270, in load
return Unpickler(file, ignore=ignore, **kwds).load()
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 472, in load
obj = StockUnpickler.load(self)
ValueError: unsupported pickle protocol: 5
[2021-08-31 16:20:30,634] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 19
[2021-08-31 16:20:30,970] {process_utils.py:66} INFO - Process psutil.Process(pid=19, status='terminated', exitcode=0, started='16:20:28') (19) terminated with exit code 0
[2021-08-31 16:20:30,970] {scheduler_job.py:1326} INFO - Exited execute loop
Traceback (most recent call last):
File "/home/airflow/.local/bin/airflow", line 8, in
sys.exit(main())
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/main.py", line 40, in main
args.func(args)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/cli/cli_parser.py", line 48, in command
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/cli.py", line 91, in wrapper
return f(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/cli/commands/scheduler_command.py", line 64, in scheduler
job.run()
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/base_job.py", line 245, in run
self._execute()
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1299, in _execute
self._run_scheduler_loop()
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1392, in _run_scheduler_loop
num_queued_tis = self._do_scheduling(session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1531, in _do_scheduling
self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1759, in _schedule_dag_run
self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1785, in verify_integrity_if_dag_changed
dag_run.verify_integrity(session=session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 638, in verify_integrity
tis = self.get_task_instances(session=session)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 328, in get_task_instances
return tis.all()
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3373, in all
return list(self)
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 100, in instances
cursor.close()
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in exit
with_traceback=exc_tb,
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise
raise exception
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in instances
rows = [proc(row) for row in fetch]
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in
rows = [proc(row) for row in fetch]
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 588, in _instance
populators,
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 725, in populate_full
dict[key] = getter(row)
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process
return loads(value)
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 275, in loads
return load(file, ignore, **kwds)
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 270, in load
return Unpickler(file, ignore=ignore, **kwds).load()
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 472, in load
obj = StockUnpickler.load(self)
ValueError: unsupported pickle protocol: 5

What you expected to happen

No response

How to reproduce

No response

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

[potiuk]

You must have a different version of python in the image installed in the original chart and the new one.

This is a duplicate of this - and you will find which tables to clean from a comment there. #14134

[kn6405]

I wonder if there are other tables with pickled values other than the two mentioned in #14134. I emptied the dag_pickle table and I cleared the conf column values in dag_run. Restarted all pod and had same issue. Even uninstalled and reinstalled the chart to see if the start job running might help.

[potiuk]

Did you take a backup of the DB before migrating? I am afraid the easiest way is to restore it and use the image with the same python version? If not then simply resetting the DB might be best way to rectify the mistake.