Airflow DAG报错：GlueJobOperator不可JSON序列化，需保留@task语法

问题根源 #

你遇到的Object of type GlueJobOperator is not JSON serializable错误，本质是**@task装饰的函数默认会将返回值存入XCom，但GlueJobOperator是复杂的Airflow操作符对象，无法被JSON序列化**。你之前的写法是把Operator实例作为任务返回值，这不符合@task的设计逻辑——@task封装的是任务执行逻辑，而非返回另一个任务操作符。

解决方案 #

以下两种方案都能保留任务映射（expand）功能，同时解决序列化问题：

方案1：@task + GlueJobHook触发任务 #

在@task函数内部使用GlueJobHook直接触发Glue任务，返回可序列化的简单类型（如任务ID），避免返回Operator实例。

示例代码：

from airflow.decorators import task, dag
from airflow.providers.amazon.aws.hooks.glue import GlueJobHook
from datetime import datetime

@task
def run_glue_job(job_name):
    # 初始化Glue Hook
    glue_hook = GlueJobHook(aws_conn_id="aws_default")
    # 触发Glue任务，获取任务运行ID
    job_run_id = glue_hook.start_job_run(job_name=job_name)
    # 可选：等待任务完成（根据业务需求决定是否添加）
    # glue_hook.wait_for_job_run(job_run_id=job_run_id, poll_interval=30)
    # 返回可序列化的任务ID，方便后续任务调用（如果需要）
    return job_run_id

@dag(start_date=datetime(2024, 8, 7), schedule=None, dag_id="glue_dynamic_dag")
def glue_dag():
    # 使用expand实现多任务映射
    run_glue_job.expand(job_name=["glue_job_01", "glue_job_02"])

glue_dag()

方案2：直接对GlueJobOperator使用expand映射 #

如果不需要在任务中添加额外逻辑，可跳过@task，直接对GlueJobOperator使用partial+expand实现动态任务映射，这种方式更简洁，且不会涉及XCom序列化问题。

示例代码：

from airflow import DAG
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from datetime import datetime

with DAG(
    start_date=datetime(2024, 8, 7),
    schedule=None,
    dag_id="glue_dynamic_dag"
) as dag:
    # 使用partial固定公共参数，expand动态传入job_name列表
    GlueJobOperator.partial(
        task_id="run_glue_job",
        aws_conn_id="aws_default"
    ).expand(job_name=["glue_job_01", "glue_job_02"])

方案选择建议 #

• 若需要在任务执行前后添加自定义逻辑（如参数校验、日志增强、结果处理），优先选择方案1。
• 若仅需简单触发多个Glue任务，方案2更轻量化，无需额外封装@task。

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