Запуск работы на воздушном потоке на основе Webrequest

Question

Я хотел знать, могут ли выполняться задачи воздушного потока при получении запроса через HTTP. Меня не интересует часть планирования Airflow. Я просто хочу использовать его в качестве заменителя сельдерея.

Таким образом, пример операции будет примерно таким.

1. Пользователь отправляет форму, требующую отчета.
2. Бэкэнд получает запрос и отправляет пользователю уведомление о том, что запрос был получен.
3. Бэкэнд затем запускает работу, используя Airflow для немедленного запуска.
4. Airflow затем выполняет серию задач, связанных с DAG. Например, сначала извлекайте данные из красного смещения, извлекайте данные из MySQL, делайте некоторые операции над двумя наборами результатов, объединяйте их, а затем загружайте результаты в Amazon S3, отправляйте электронное письмо.

Из всего, что я читал в Интернете, вы можете запускать задания воздушного потока, выполняя airflow ... в командной строке. Мне было интересно, есть ли api python, который может выполнять одно и то же.

Спасибо.

Answer 1

Airflow REST API Plugin поможет вам здесь. После того, как вы выполнили инструкции по установке плагина, вам просто нужно нажать следующий URL-адрес: 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, либо пропустив run_id, либо указав уникальный идентификатор, и передав кодировку json для conf (с любой из параметров, которые вам нужны в запущенном dag).

Вот пример функции JavaScript, который использует JQuery, чтобы вызвать воздушный поток апи:

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

Вы должны посмотреть на Airflow HTTP Sensor для ваших нужд. Вы можете использовать это, чтобы вызвать даг.

Answer 3

Новая опция в потоке воздуха является экспериментальной, но встроенный, API конечных точек в более поздних сборках 1,7 и 1,8. Это позволяет вам запускать службу REST на вашем сервере воздушного потока, чтобы прослушивать порт и принимать задания cli.

У меня только ограниченный опыт, но я успешно прошел тест-пробку. В документах:

/api/experimental/dags/<DAG_ID>/dag_runs создает dag_run для заданного идентификатора dag (POST).

Это будет запланировать немедленный запуск любого трюка, который вы хотите запустить. Тем не менее, он все еще использует планировщик, ожидая, когда будет слышен пульс, чтобы он запустил и передал задания работнику. Это точно такое же поведение, что и CLI, поэтому я по-прежнему считаю, что он подходит для вашего прецедента.

Документация о том, как настроить его можно здесь: https://airflow.apache.org/api.html

Есть несколько простых примеров клиентов в GitHub, тоже под вентиляционными/API/клиентов