GCP Composer中BashOperator执行Dataproc流任务15小时后失败求助

核心原因：Airflow任务超时限制 #

Airflow的Executor（比如CeleryExecutor、KubernetesExecutor）默认有任务执行超时阈值，大多在15-16小时左右。你的BashOperator会一直等待流式任务的启动脚本执行完成，但流式任务是持续运行不会终止的，导致BashOperator进程长期挂起，一旦超过Executor的超时时间，就会被强制终止，Airflow随即标记任务失败。而Dataproc任务已经独立启动，不受Airflow进程终止的影响，所以会继续正常运行。

其他可能的次要原因 #

• 网络波动：Airflow Worker和Dataproc集群之间的临时网络中断，导致BashOperator的输出流中断，被Airflow判定为任务失败。
• Worker资源耗尽：长时间运行的BashOperator进程占用Worker节点的内存、文件句柄等资源，达到系统限制后被Kill，进而触发任务失败标记。

方案1：让启动脚本后台运行，BashOperator快速退出 #

修改bash_command，把启动命令放到后台执行并重定向输出，这样BashOperator在启动Dataproc任务后会立即退出，Airflow直接标记任务成功，从根源避免超时问题。示例代码：

sparkstreaming = BashOperator(
    task_id='sparkstreaming',
    retries=0,
    bash_command= f'gsutil cp gs://bkt-gcp-spark/start-spark.sh . && nohup bash start-spark.sh > /dev/null 2>&1 &',
    dag=dag
)

• nohup：让脚本脱离终端，即使Airflow进程退出也不会影响Dataproc任务
• > /dev/null 2>&1：把所有标准输出和错误输出重定向到空设备，避免进程因输出缓冲区满而挂起
• &：让脚本在后台执行，BashOperator执行完命令就会结束

方案2：延长Airflow任务超时时间（不推荐） #

如果一定要让BashOperator持续挂着，可以修改Airflow的超时配置，延长阈值。比如在BashOperator中单独设置：

from datetime import timedelta

sparkstreaming = BashOperator(
    task_id='sparkstreaming',
    retries=0,
    bash_command= f'gsutil cp gs://bkt-gcp-spark/start-spark.sh . && bash start-spark.sh',
    execution_timeout=timedelta(hours=48),  # 设置48小时超时
    dag=dag
)

或者修改airflow.cfg中的全局超时参数（比如CeleryExecutor的task_time_limit）。但这种方法只是推迟超时，流式任务永远不会结束，最终还是会触发失败标记，不是长久之计。

方案3：使用官方DataprocOperator替代BashOperator #

Airflow提供了专门的DataprocSubmitJobOperator，专门用于提交Dataproc任务，提交成功后立即标记任务完成，不需要等待任务结束，完美适配流式场景。示例代码：

from airflow.providers.google.cloud.operators.dataproc import DataprocSubmitJobOperator

spark_stream_job = {
    "reference": {"project_id": "你的GCP项目ID"},
    "placement": {"cluster_name": "你的Dataproc集群名"},
    "spark_job": {
        "main_class": "你的流式任务主类",
        "jar_file_uris": ["gs://bkt-gcp-spark/你的流式任务Jar包路径"],
        "args": ["任务参数1", "任务参数2"]  # 可选，根据你的任务需求添加
    }
}

sparkstreaming = DataprocSubmitJobOperator(
    task_id="sparkstreaming",
    job=spark_stream_job,
    region="你的GCP区域",
    project_id="你的GCP项目ID",
    dag=dag
)

优势：

• 原生适配GCP Dataproc，无需手动处理脚本复制、后台运行等操作
• 可以搭配DataprocJobSensor监控任务运行状态（可选）
• 从根本上避免BashOperator的超时问题

方案4：给启动脚本加心跳机制（应急方案） #

如果暂时无法替换BashOperator，可以在start-spark.sh中添加心跳循环，定期输出内容保持进程活跃，避免被超时终止：

# start-spark.sh内容
# 启动Dataproc流式任务
gcloud dataproc jobs submit spark --cluster=你的集群名 --jar=gs://bkt-gcp-spark/你的Jar包 &
JOB_PID=$!

# 每隔1小时输出一次心跳，保持进程活跃
while kill -0 $JOB_PID 2>/dev/null; do
    echo "Dataproc流式任务运行中..."
    sleep 3600
done

不过这种方法还是依赖超时配置，只是延长了触发失败的时间，不如方案1或3彻底。

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