Repository navigation
AttributeError: Can't pickle local object 'DaskExecutor.execute_async.<locals>.airflow_run' #11451
Description
Apache Airflow version: 1.10.12
Kubernetes version (if you are using kubernetes) (use kubectl version):
Environment:
- OS (e.g. from /etc/os-release): centos7.4
What happened:
DaskExecutor will raise ValueError: Cell is empty error when scheduling task. no dag can be scheduled. debug list below:
-bash-4.2$ airflow scheduler
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ \_ | /| / /
___ ___ | / _ / _ __/ _ / / /_/ /_ |/ |/ /
_/_/ |_/_/ /_/ /_/ /_/ \____/____/|__/
[2020-07-16 18:48:30,885] {__init__.py:51} INFO - Using executor DaskExecutor
[2020-07-16 18:48:30,890] {scheduler_job.py:1346} INFO - Starting the scheduler
[2020-07-16 18:48:30,890] {scheduler_job.py:1354} INFO - Running execute loop for -1 seconds
[2020-07-16 18:48:30,890] {scheduler_job.py:1355} INFO - Processing each file at most -1 times
[2020-07-16 18:48:30,890] {scheduler_job.py:1358} INFO - Searching for files in /ansdep/opsapi/dags
[2020-07-16 18:48:30,890] {scheduler_job.py:1360} INFO - There are 0 files in /ansdep/opsapi/dags
[2020-07-16 18:48:30,898] {scheduler_job.py:1411} INFO - Resetting orphaned tasks for active dag runs
[2020-07-16 18:48:30,912] {dag_processing.py:556} INFO - Launched DagFileProcessorManager with pid: 70385
[2020-07-16 18:48:30,916] {settings.py:54} INFO - Configured default timezone <Timezone [Asia/Shanghai]>
[2020-07-16 18:49:22,984] {scheduler_job.py:951} INFO - 1 tasks up for execution:
<TaskInstance: bc081861-e948-4e59-92fa-fc14f4330eb2.first_for_flush_state 2020-07-16 10:49:00+00:00 [scheduled]>
[2020-07-16 18:49:22,990] {scheduler_job.py:982} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued
[2020-07-16 18:49:22,990] {scheduler_job.py:1010} INFO - DAG bc081861-e948-4e59-92fa-fc14f4330eb2 has 0/512 running and queued tasks
[2020-07-16 18:49:22,994] {scheduler_job.py:1060} INFO - Setting the following tasks to queued state:
<TaskInstance: bc081861-e948-4e59-92fa-fc14f4330eb2.first_for_flush_state 2020-07-16 10:49:00+00:00 [scheduled]>
[2020-07-16 18:49:23,000] {scheduler_job.py:1134} INFO - Setting the following 1 tasks to queued state:
<TaskInstance: bc081861-e948-4e59-92fa-fc14f4330eb2.first_for_flush_state 2020-07-16 10:49:00+00:00 [queued]>
[2020-07-16 18:49:23,000] {scheduler_job.py:1170} INFO - Sending ('bc081861-e948-4e59-92fa-fc14f4330eb2', 'first_for_flush_state', datetime.datetime(2020, 7, 16, 10, 49, tzinfo=<TimezoneInfo [UTC, GMT, +00:00:00, STD]>), 1) to executor with priority 4 and queue default
[2020-07-16 18:49:23,000] {base_executor.py:58} INFO - Adding to queue: ['airflow', 'run', 'bc081861-e948-4e59-92fa-fc14f4330eb2', 'first_for_flush_state', '2020-07-16T10:49:00+00:00', '--local', '--pool', 'default_pool', '-sd', '/ansdep/opsapi/dags/bc081861-e948-4e59-92fa-fc14f4330eb2.py']
/ansdep/python3/lib/python3.7/site-packages/airflow/executors/dask_executor.py:63: UserWarning: DaskExecutor does not support queues. All tasks will be run in the same cluster
'DaskExecutor does not support queues. '
distributed.protocol.pickle - INFO - Failed to serialize <function DaskExecutor.execute_async.<locals>.airflow_run at 0x7f748ce808c8>. Exception: Cell is empty
[2020-07-16 18:49:23,003] {scheduler_job.py:1384} ERROR - Exception when executing execute_helper
Traceback (most recent call last):
File "/ansdep/python3/lib/python3.7/site-packages/distributed/worker.py", line 3323, in dumps_function
result = cache_dumps[func]
File "/ansdep/python3/lib/python3.7/site-packages/distributed/utils.py", line 1549, in __getitem__
value = super().__getitem__(key)
File "/ansdep/python3/lib/python3.7/collections/__init__.py", line 1025, in __getitem__
raise KeyError(key)
KeyError: <function DaskExecutor.execute_async.<locals>.airflow_run at 0x7f748ce808c8>
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/ansdep/python3/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 41, in dumps
result = pickle.dumps(x, **dump_kwargs)
AttributeError: Can't pickle local object 'DaskExecutor.execute_async.<locals>.airflow_run'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/ansdep/python3/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1382, in _execute
self._execute_helper()
File "/ansdep/python3/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1443, in _execute_helper
if not self._validate_and_run_task_instances(simple_dag_bag=simple_dag_bag):
File "/ansdep/python3/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1505, in _validate_and_run_task_instances
self.executor.heartbeat()
File "/ansdep/python3/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 130, in heartbeat
self.trigger_tasks(open_slots)
File "/ansdep/python3/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 154, in trigger_tasks
executor_config=simple_ti.executor_config)
File "/ansdep/python3/lib/python3.7/site-packages/airflow/executors/dask_executor.py", line 70, in execute_async
future = self.client.submit(airflow_run, pure=False)
File "/ansdep/python3/lib/python3.7/site-packages/distributed/client.py", line 1579, in submit
actors=actor,
File "/ansdep/python3/lib/python3.7/site-packages/distributed/client.py", line 2598, in _graph_to_futures
"tasks": valmap(dumps_task, dsk3),
File "/ansdep/python3/lib/python3.7/site-packages/toolz/dicttoolz.py", line 83, in valmap
rv.update(zip(iterkeys(d), map(func, itervalues(d))))
File "/ansdep/python3/lib/python3.7/site-packages/distributed/worker.py", line 3361, in dumps_task
return {"function": dumps_function(task[0]), "args": warn_dumps(task[1:])}
File "/ansdep/python3/lib/python3.7/site-packages/distributed/worker.py", line 3325, in dumps_function
result = pickle.dumps(func)
File "/ansdep/python3/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 52, in dumps
result = cloudpickle.dumps(x, **dump_kwargs)
File "/ansdep/python3/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 101, in dumps
cp.dump(obj)
File "/ansdep/python3/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 540, in dump
return Pickler.dump(self, obj)
File "/ansdep/python3/lib/python3.7/pickle.py", line 437, in dump
self.save(obj)
File "/ansdep/python3/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/ansdep/python3/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 722, in save_function
*self._dynamic_function_reduce(obj), obj=obj
File "/ansdep/python3/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 659, in _save_reduce_pickle5
dictitems=dictitems, obj=obj
File "/ansdep/python3/lib/python3.7/pickle.py", line 638, in save_reduce
save(args)
File "/ansdep/python3/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/ansdep/python3/lib/python3.7/pickle.py", line 786, in save_tuple
save(element)
File "/ansdep/python3/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/ansdep/python3/lib/python3.7/pickle.py", line 771, in save_tuple
save(element)
File "/ansdep/python3/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/ansdep/python3/lib/python3.7/site-packages/dill/_dill.py", line 1169, in save_cell
f = obj.cell_contents
ValueError: Cell is empty
[2020-07-16 18:49:23,014] {helpers.py:325} INFO - Sending Signals.SIGTERM to GPID 70385
[2020-07-16 18:49:23,041] {helpers.py:291} INFO - Process psutil.Process(pid=70385, status='terminated') (70385) terminated with exit code 0
[2020-07-16 18:49:23,041] {helpers.py:291} INFO - Process psutil.Process(pid=70575, status='terminated') (70575) terminated with exit code None
[2020-07-16 18:49:23,042] {scheduler_job.py:1387} INFO - Exited execute loop
How to reproduce it:
keep argument executor equal to DaskExecutor. startup services dask-scheduler/dask-worker/airflow webserver/airflow scheduler and trigger any example dag in website.
# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor, KubernetesExecutor
executor = DaskExecutor
Anything else we need to know:
it seems that the exception raised when picker serialize the airflow_run method defined in airflow.executor.dask_executor.DaskExecutor.execute_async
def execute_async(self, key, command, queue=None, executor_config=None):
if queue is not None:
warnings.warn(
'DaskExecutor does not support queues. '
'All tasks will be run in the same cluster'
)
if command[0:2] != ["airflow", "run"]:
raise ValueError('The command must start with ["airflow", "run"].')
def airflow_run():
return subprocess.check_call(command, close_fds=True)
future = self.client.submit(airflow_run, pure=False) < -- airflow_run will be pickle serialized and submit to dask-scheduler
self.futures[future] = key
picker seem not enable dumps a function defined in function. so, i change the code and scheduler work well:
def airflow_run(command):
return subprocess.check_call(command, close_fds=True)
class DaskExecutor(BaseExecutor):
...
def execute_async(self, key, command, queue=None, executor_config=None):
if queue is not None:
warnings.warn(
'DaskExecutor does not support queues. '
'All tasks will be run in the same cluster'
)
if command[0:2] != ["airflow", "run"]:
raise ValueError('The command must start with ["airflow", "run"].')
future = self.client.submit(airflow_run, command, pure=False)
self.futures[future] = key
Activity
boring-cyborg commentedon Oct 12, 2020
Thanks for opening your first issue here! Be sure to follow the issue template!
lafrinte commentedon Oct 21, 2020
i modify the description and follow the template. wish your replay
toughrogrammer commentedon Jan 1, 2021
I got similar error when I upgraded to 2.0.0
xawyong commentedon Jan 7, 2021
use 2.0.0, after changed the code as @lafrinte proposed, got below error.
griffinqiu commentedon Jan 19, 2021
Same issue
esc commentedon Mar 16, 2021
I had the same issue and this fixed it for me:
uqfoundation/dill#383 (comment)