Airflow DAG：我的执行在第一个任务后“终止”

Question

我尝试了一个简单的 DAG 来运行 3 个任务，每个任务上都有传感器，在 EMR 上运行，如下所示：

EMR_creation -> T1 -> S1 -> T2 -> S2 -> T3 -> S3 -> EMR_termination.

我遇到的问题是 T1-S1 成功了，但由于错误，T2 从未启动：

错误 - 调用 AddJobFlowSteps 操作时发生错误 (ValidationException)：可能无法修改正在关闭、终止或完成的作业流。

当我设置重试步骤而不在 EMR 集群中记录时，同样的错误会重复，这是有道理的。我曾尝试将不同的任务放在 T1 或 T2 中，但没有改进。

Answer 1

假设 EMR 在没有步骤的情况下终止您的工作流程

有一个名为KeepJobFlowAliveWhenNoSteps的参数可以在所有步骤完成后停止终止EMR作业流程，您可以使用它。见boto3 docs。

如果 JobFlowInstancesConfig KeepJobFlowAliveWhenNoSteps 参数 设置为TRUE，集群转换到WAITING状态而不是 而不是在步骤完成后关闭。

对于气流，它可以在 emr_connection 或您使用的 job_flow_overrides 字典中。

例如emr_connection:

{
  "LogUri": "s3n://aws-logs-***-us-east-1/elasticmapreduce/",
  "ReleaseLabel": "emr-5.28.0",
  "Instances": {
    "InstanceGroups": ["..."],
    "KeepJobFlowAliveWhenNoSteps": true,  -- here!
    "Ec2KeyName": "emr-key"
  },
  "VisibleToAllUsers": true,
  "BootstrapActions": ["..."],
  "JobFlowRole": "???",
  "ServiceRole": "??"
}