如何使用Airflow重新启动失败的结构化流spark作业？

Question

我需要在AWS EMR中运行结构化流spark作业。作为弹性需求，如果spark作业由于某些原因而失败，我们希望spark作业可以在EMR中重新创建。类似于ECS中的任务编排，如果健康检查失败，可以重启任务。然而，EMR更多是一个计算引擎，而不是编排系统。

我正在寻找一些大数据工作流程编排工具，例如Airflow。但是，它不能支持DAG中的循环。如何实现如下所示的一些功能？

step_adder (EmrAddStepsOperator) >> step_checker (EmrStepSensor) >> step_adder (EmrAddStepsOperator)。

建议的方法是什么来提高这种工作水平的弹性？欢迎提出任何意见！

Answer 1

Apache Spark已经涵盖了一些弹性(使用Spark-submit提交的作业)，但是，当您想要与不同的进程交互时，不使用Spark，那么Airflow可能是一个解决方案。在您的示例中，Sensor可以帮助检测是否发生了特定条件。基于此，您可以在DAG中做出决定。下面是一个简单的HttpSensor，它等待一个批处理作业，看看它是否成功完成

wait_batch_to_finish = HttpSensor(
    http_conn_id='spark_web',
    task_id="wait_batch_to_finish",
    method="GET",
    headers={"Content-Type": "application/json"},
    endpoint="/json",
    response_check=lambda response: check_spark_status(response, "{{ ti.xcom_pull('batch_intel_task')}}"),
    poke_interval=60,
    dag=dag
)