基于Webrequest在Airflow上运行作业

Question

我想知道在通过HTTP获取请求时是否可以执行气流任务.我对Airflow的调度部分不感兴趣.我只想用它作为芹菜的替代品.

所以一个示例操作就是这样的.

1. 用户提交请求报告的表单.
2. 后端接收请求并向用户发送已收到请求的通知.
3. 然后,后端使用Airflow安排作业立即运行.
4. 然后,Airflow执行与DAG相关的一系列任务.例如,首先从redshift中提取数据,从MySQL提取数据,在两个结果集上进行一些操作,将它们组合,然后将结果上传到Amazon S3,发送电子邮件.

从我在线阅读的内容中,您可以通过airflow ...在命令行上执行来运行气流作业.我想知道是否有一个python api可以执行相同的事情.

谢谢.

Answer 1

该气流REST API插件将帮助你在这里.一旦你按照安装插件的说明,你只需要点击以下网址:http://{HOST}:{PORT}/admin/rest_api/api/v1.0/trigger_dag?dag_id={dag_id}&run_id={run_id}&conf={url_encoded_json_parameters},用你的dag的id替换dag_id,省略run_id或指定一个唯一的id,并传递一个url编码的json用于conf(任何您在触发的dag中需要的参数).

这是一个使用jQuery调用Airflow api的JavaScript函数示例:

function triggerDag(dagId, dagParameters){
    var urlEncodedParameters = encodeURIComponent(dagParameters);
    var dagRunUrl = "http://airflow:8080/admin/rest_api/api/v1.0/trigger_dag?dag_id="+dagId+"&conf="+urlEncodedParameters;
    $.ajax({
        url: dagRunUrl,
        dataType: "json",
        success: function(msg) {
            console.log('Successfully started the dag');
        },
        error: function(e){
           console.log('Failed to start the dag');
        }
    });
}

Answer 2

气流中的一个新选项是最新版本1.7和1.8中的实验性内置API端点.这允许您在气流服务器上运行REST服务以侦听端口并接受cli作业.

我自己只有有限的经验,但我已成功运行测试骰子.根据文档:

/api/experimental/dags/<DAG_ID>/dag_runs 为给定的dag id(POST)创建一个dag_run.

这将安排你想要运行的任何dag的立即运行.但它仍然使用调度程序,等待心跳看到dag正在运行并将任务传递给worker.这与CLI的行为完全相同,所以我仍然认为它适合您的用例.

有关如何配置它的文档,请访问:https://airflow.apache.org/api.html

在airflow/api/clients下,github中也有一些简单的示例客户端

Answer 3

您应该根据需要查看Airflow HTTP Sensor.您可以使用它来触发dag.