Avi*_*ava 5 python celery python-3.x kubernetes airflow
我使用Celery Executor有多个dag,但是我想使用Kubernetes Executor运行一个特定的dag。我无法推断出实现此目标的好方法。
我已经airflow.cfg声明CeleryExecutor要在其中使用。而且我不想更改它,因为除了一只狗以外,其他所有狗狗都确实需要它。
# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor
executor = CeleryExecutor
Run Code Online (Sandbox Code Playgroud)
我的问题代码:
from datetime import datetime, timedelta
from airflow import DAG
from airflow.contrib.operators.kubernetes_pod_operator import \
KubernetesPodOperator
from airflow.operators.dummy_operator import DummyOperator
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime.utcnow(),
'email': ['airflow@example.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5)
}
dag = DAG(
'kubernetes_sample_1', default_args=default_args)
start = DummyOperator(task_id='run_this_first', dag=dag)
passing = KubernetesPodOperator(namespace='default',
image="Python:3.6",
cmds=["Python", "-c"],
arguments=["print('hello world')"],
labels={"foo": "bar"},
name="passing-test",
task_id="passing-task",
get_logs=True,
dag=dag
)
failing = KubernetesPodOperator(namespace='default',
image="ubuntu:1604",
cmds=["Python", "-c"],
arguments=["print('hello world')"],
labels={"foo": "bar"},
name="fail",
task_id="failing-task",
get_logs=True,
dag=dag
)
passing.set_upstream(start)
failing.set_upstream(start)
Run Code Online (Sandbox Code Playgroud)
我可以设置if-else条件,然后从Airflow选择配置的位置更改该值。如果听起来不错,请告诉我路径和文件。尽管我希望获得一种更成熟的方法,如果可以的话。
小智 0
我认为不可能同时使用两个执行器。但是您可以只使用 CeleryExecutor,但使用 KubernetesPodOperator 声明资源密集型任务,解决问题的作业由 CeleryExecutor 调度/监视,并由 Kubernetes 运行以实现任务中的实际处理逻辑。
| 归档时间: |
|
| 查看次数: |
175 次 |
| 最近记录: |