Airflow Kubernetes Executor 可以运行任何 Airflow Operator,只需要将 Operator 实例添加到 DAG 的任务中。
下面是一个示例,演示如何在 Airflow Kubernetes Executor 中运行 BashOperator:
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from datetime import datetime
default_args = {
'owner': 'airflow',
'start_date': datetime(2021, 1, 1)
}
dag = DAG('kubernetes_executor_example', default_args=default_args, schedule_interval=None)
# 使用 BashOperator 在 Kubernetes 中运行命令
task1 = BashOperator(
task_id='task1',
bash_command='echo "Hello from BashOperator"',
dag=dag
)
# 使用 KubernetesPodOperator 在 Kubernetes 中运行容器
task2 = KubernetesPodOperator(
task_id='task2',
name='task2',
image='ubuntu:latest',
cmds=['echo', 'Hello from KubernetesPodOperator'],
dag=dag
)
# 设置任务之间的依赖关系
task1 >> task2
在上面的示例中,我们使用了 BashOperator 和 KubernetesPodOperator。BashOperator 用于在 Kubernetes 中运行 Bash 命令,而 KubernetesPodOperator 用于在 Kubernetes 中运行容器。这些 Operator 可以与其他 Airflow Operator 一起使用,以构建完整的任务流程。