我想配置 Dask 以在 Airflow 中分发 DAG。 我读过https://airflow.apache.org/howto/executor/use-dask.html和 https://distributed.readthedocs.io/en/latest/ ，但我不明白它是如何工作的。 我有两台 Apache Aiflow 服务器，在哪里运行 dask-shedulers 和 dask-workers，当第一台服务器掉线时，第二台服务器自动运行？ 据我了解，Airflow 将任务放在 dask-sheduler 中。我不明白如何在两台服务器上交 friend 。我不明白为什么需要 dask-workers 以及它的作用，我认为这是某种不必要的组件。
我不想使用 CeleryExecutor 并为 Celery 配置 RabbitMQ 或 Redis
请您参考如下方法:
您可以在此处找到两个更好的描述来启动和运行 Airflow + Dask:https://www.alibabacloud.com/blog/schedule-data-lake-analytics-tasks-by-using-airflow_594183或 https://tech.marksblogg.com/install-and-configure-apache-airflow.html .特别是第一个链接对我帮助很大。而且，运行 Airflow scheduler + webserver + dask-scheduler 就可以了。 但是，一旦我启动 Airflow worker + dask-worker，Airflow worker 就会退出并提示没有激活 Celery:
ModuleNotFoundError: No module named 'celery'
当我在没有 Airflow worker 的情况下运行 dask-worker 时，一切似乎都正常，直到我触发 DAG:
worker_1 | [2019-05-12 20:47:05,527] {__init__.py:51} INFO - Using executor DaskExecutor
worker_1 | usage: airflow [-h]
worker_1 | {backfill,list_dag_runs,list_tasks,clear,pause,unpause,trigger_dag,delete_dag,pool,variables,kerberos,render,run,initdb,list_dags,dag_state,task_failed_deps,task_state,serve_logs,test,webserver,resetdb,upgradedb,scheduler,worker,flower,version,connections,create_user,delete_user,list_users,sync_perm,next_execution,rotate_fernet_key}
worker_1 | ...
worker_1 | airflow: error: the following arguments are required: subcommand
worker_1 | distributed.worker - WARNING - Compute Failed
worker_1 | Function: airflow_run
worker_1 | args: ()
worker_1 | kwargs: {}
worker_1 | Exception: CalledProcessError(2, ['airflow', 'run', 'example_python_operator', 'print_the_context', '2019-05-12T20:47:02.111022+00:00', '--pickle', '13', '--local', '-sd', '/opt/airflow/dags/python_exec.py'])
worker_1 |
webserver_1 | [2019-05-12 20:47:06 +0000] [37] [INFO] Handling signal: ttin
webserver_1 | [2019-05-12 20:47:06 +0000] [744] [INFO] Booting worker with pid: 744
webserver_1 | [2019-05-12 20:47:06,299] {dask_executor.py:77} ERROR - Failed to execute task: CalledProcessError(2, ['airflow', 'run', 'example_python_operator', 'print_the_context', '2019-05-12T20:47:02.111022+00:00', '--pickle', '13', '--local', '-sd', '/opt/airflow/dags/python_exec.py'])
有什么提示可以解决这个问题吗？
