Airflow Worker not Working When I run Backfill commands

[r39132]

I am using the CeleryExecutor with a Postgres results backend and queue. It looks like this fails when I run "airflow backfill" commands but works when I use the UI.

Sid-As-MBP-15:airflow siddharth$ airflow worker

 -------------- celery@Sid-As-MacBook-Pro-15.local v3.1.18 (Cipater)
---- **** ----- 
--- * ***  * -- Darwin-14.3.0-x86_64-i386-64bit
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         airflow.executors.celery_executor:0x10c04e690
- ** ---------- .> transport:   sqla+postgresql://siddharth@localhost:5432/airflow_celery
- ** ---------- .> results:     db+postgresql://siddharth@localhost:5432/airflow_celery
- *** --- * --- .> concurrency: 16 (prefork)
-- ******* ---- 
--- ***** ----- [queues]
 -------------- .> default          exchange=celery(direct) key=celery


[2015-06-22 15:08:15,337: WARNING/MainProcess] celery@Sid-As-MacBook-Pro-15.local ready.
Starting flask
 * Running on http://0.0.0.0:8793/ (Press CTRL+C to quit)
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 10, in <module>
    args.func(args)
  File "/usr/local/lib/python2.7/site-packages/airflow/bin/cli.py", line 97, in run
    DagPickle).filter(DagPickle.id == args.pickle).first()
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2444, in first
    ret = list(self[0:1])
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/orm/query.py", line 2280, in __getitem__
    return list(res)
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/orm/loading.py", line 84, in instances
    util.raise_from_cause(err)
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/util/compat.py", line 199, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb)
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/orm/loading.py", line 69, in instances
    rows = [proc(row) for row in fetch]
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/orm/loading.py", line 426, in _instance
    loaded_instance, populate_existing, populators)
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/orm/loading.py", line 484, in _populate_full
    dict_[key] = getter(row)
  File "/usr/local/lib/python2.7/site-packages/sqlalchemy/sql/sqltypes.py", line 1261, in process
    return loads(value)
  File "/usr/local/lib/python2.7/site-packages/dill/dill.py", line 209, in loads
    return load(file)
  File "/usr/local/lib/python2.7/site-packages/dill/dill.py", line 199, in load
    obj = pik.load()
  File "/usr/local/Cellar/python/2.7.9/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 858, in load
    dispatch[key](self)
  File "/usr/local/Cellar/python/2.7.9/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 1090, in load_global
    klass = self.find_class(module, name)
  File "/usr/local/lib/python2.7/site-packages/dill/dill.py", line 278, in find_class
    return StockUnpickler.find_class(self, module, name)
  File "/usr/local/Cellar/python/2.7.9/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 1124, in find_class
    __import__(module)
ImportError: No module named python_operator
[2015-06-22 15:11:08,787: ERROR/Worker-14] 1
[2015-06-22 15:11:08,835: ERROR/MainProcess] Task airflow.executors.celery_executor.execute_command[4a012cf5-83f6-4d54-af34-b2ce74b517db] raised unexpected: Exception('Celery command failed',)
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/usr/local/lib/python2.7/site-packages/celery/app/trace.py", line 438, in __protected_call__
    return self.run(*args, **kwargs)
  File "/usr/local/lib/python2.7/site-packages/airflow/executors/celery_executor.py", line 41, in execute_command
    raise Exception('Celery command failed')
Exception: Celery command failed

My airflow.cfg file has the following settings for celery

# The Celery broker URL. Celery supports RabbitMQ, Redis and experimentaly
# a sqlalchemy database. Refer to the Celery documentation for more
# information.
#broker_url = sqla+mysql://airflow:airflow@localhost:3306/airflow
broker_url = sqla+postgresql://siddharth@localhost:5432/airflow_celery

# Another key Celery setting
#celery_result_backend = db+mysql://airflow:airflow@localhost:3306/airflow
celery_result_backend = db+postgresql://siddharth@localhost:5432/airflow_celery

I am getting this error with the example_python_operator and example_bash_operator flows. I haven't tried others.

airflow flower is not much help for experimental brokers like Postgres (over SQLAlchemy) since many commands are unsupported as mentioned in the Limitations section of this

Sid-As-MacBook-Pro-15:airflow siddharth$ airflow flower
[I 150622 15:05:54 command:114] Visit me at http://localhost:5555
[I 150622 15:05:55 command:116] Broker: sqla+postgresql://siddharth@localhost:5432/airflow_celery
[I 150622 15:05:55 command:119] Registered tasks:
    ['celery.backend_cleanup',
     'celery.chain',
     'celery.chord',
     'celery.chord_unlock',
     'celery.chunks',
     'celery.group',
     'celery.map',
     'celery.starmap']
[I 150622 15:05:55 mixins:225] Connected to sqla+postgresql://siddharth@localhost:5432/airflow_celery
[W 150622 15:05:58 control:43] 'stats' inspect method failed
[W 150622 15:05:58 control:43] 'active_queues' inspect method failed
[W 150622 15:05:58 control:43] 'registered' inspect method failed
[W 150622 15:05:58 control:43] 'scheduled' inspect method failed
[W 150622 15:05:58 control:43] 'active' inspect method failed
[W 150622 15:05:58 control:43] 'reserved' inspect method failed
[W 150622 15:05:58 control:43] 'revoked' inspect method failed
[W 150622 15:05:58 control:43] 'conf' inspect method failed

If I look into my database (airflow_celery), I find 2 tables : kombu_message and kombu_queue.

In kombu_queue, I'm expecting a single queue called default, but I see several

airflow_celery=# select * from kombu_queue ;
 id |                           name                           
----+----------------------------------------------------------
  1 | default
  2 | celery@Sid-As-MacBook-Pro-15.local.celery.pidbox
  3 | celeryev.ad9d88a9-05b7-4d01-ad2c-8f95662787a0
  4 | 37bb18ec-a47f-349b-859f-3b454d2602a3.reply.celery.pidbox
  5 | b15e35ef-8873-31d6-8841-af833b9cec3d.reply.celery.pidbox
  6 | 4572302e-65a9-3ec1-850e-aff27cd59a9b.reply.celery.pidbox
  7 | 33d4dcc5-f997-3b7c-9f41-66d9a2c51159.reply.celery.pidbox

[mistercrunch]

The command runs as a subprocess so you have to make sure that a virgin shell can run the command.

You need to basically be able to start a brand new shell and run that command.

[r39132]

I'm using the local executor, so I am closing this for now. If I get this again at some later point, I will re-open.