使用Airflow在本地Kubernetes调度Spring Batch任务及依赖配置咨询

一、前置准备 #

• 确保Airflow已部署在本地K8s集群中（推荐用Helm部署或自定义Manifest），且启用KubernetesExecutor或搭配KubernetesPodOperator的CeleryExecutor
• 将Spring Batch任务打包成Docker镜像，推送到本地私有镜像仓库（避免依赖外网镜像源）
• 给Airflow服务账号配置K8s权限：允许创建/删除Pod、访问指定命名空间、拉取私有镜像仓库的镜像

二、Airflow基础配置 #

1. 修改airflow.cfg配置Kubernetes核心参数：

   executor = KubernetesExecutor
   kubernetes_container_image = 你的Airflow Worker镜像（KubernetesExecutor模式下必填）
   kubernetes_namespace = airflow（或自定义命名空间）
   kubernetes_in_cluster = True（Airflow部署在K8s内时启用集群内配置）
   

2. 在Airflow UI配置K8s连接：
   • 连接类型选择Kubernetes Cluster Connection
   • 集群内部署勾选In Cluster Configuration；外部部署则上传本地kubeconfig文件

三、编写Airflow DAG调度Spring Batch任务 #

创建Python DAG文件，通过KubernetesPodOperator启动Spring Batch容器：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'spring_batch_on_k8s',
    default_args=default_args,
    description='调度本地K8s集群上的Spring Batch任务',
    schedule_interval='0 0 * * *',
    catchup=False,
) as dag:

    run_spring_batch = KubernetesPodOperator(
        task_id='run_spring_batch_job',
        name='spring-batch-job-pod',
        namespace='batch-jobs',  # Spring Batch任务运行的目标命名空间
        image='本地镜像仓库地址/your-spring-batch-app:v1',  # 你的Spring Batch镜像地址
        cmds=['java'],
        arguments=['-jar', '/app/your-batch-app.jar', '--job.name=targetBatchJob'],
        resources={'request_cpu': '1', 'request_memory': '2Gi', 'limit_cpu': '2', 'limit_memory': '4Gi'},
        get_logs=True,  # 捕获容器日志到Airflow UI
        is_delete_operator_pod=True,  # 任务完成后自动清理Pod
        image_pull_secrets='your-registry-secret',  # 拉取私有镜像的密钥
        env_vars={  # 传递环境变量给Spring Batch应用
            'DB_URL': 'jdbc:mysql://db-host:3306/batch_db',
            'DB_USER': 'batch_user'
        }
    )

四、Spring Batch任务适配要点 #

• 确保应用启动命令支持通过参数指定Job（比如示例中的--job.name）
• 配置日志输出到stdout/stderr，方便Airflow捕获任务日志
• 应用退出码遵循规范：成功返回0，失败返回非0值（Airflow将根据退出码判断任务状态）

同DAG内依赖（任务在同一DAG中） #

直接用>>/<<符号定义执行顺序：

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

# 定义PySpark任务
run_pyspark_job = SparkSubmitOperator(
    task_id='run_pyspark_job',
    application='/path/to/your_pyspark_script.py',
    conn_id='spark_default',  # Airflow中配置的Spark连接ID
    executor_cores=2,
    executor_memory='4g',
    driver_memory='2g',
    name='pyspark-processing-task'
)

# 设置依赖：PySpark任务完成后执行Spring Batch任务
run_pyspark_job >> run_spring_batch

跨DAG依赖（任务分属不同DAG） #

使用TriggerDagRunOperator触发目标DAG：

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

trigger_spring_batch_dag = TriggerDagRunOperator(
    task_id='trigger_spring_batch_dag',
    trigger_dag_id='spring_batch_on_k8s',  # 目标Spring Batch DAG的ID
    wait_for_completion=True,  # 等待目标DAG执行完成
    poke_interval=60,  # 每隔60秒检查一次目标DAG状态
)

run_pyspark_job >> trigger_spring_batch_dag

内容的提问来源于stack exchange，提问作者ranjith meda