Uruchamianie zadania przy przepływie powietrza na podstawie zgłoszenia internetowego

Question

Chciałem wiedzieć, czy zadania przepływu powietrza mogą być wykonywane po otrzymaniu żądania przez HTTP. Nie interesuje mnie część planowania przepływu powietrza. Po prostu chcę go użyć jako substytutu selera.

Tak więc przykładowa operacja byłaby czymś takim.

1. Użytkownik przesyła formularz z prośbą o zgłoszenie.
2. Backend odbiera żądanie i wysyła użytkownikowi powiadomienie, że żądanie zostało odebrane.
3. Urządzenie do planowania następnie ustawia zlecenie za pomocą funkcji Airflow, aby natychmiast uruchomić.
4. Airflow następnie wykonuje serię zadań związanych z DAG. Na przykład najpierw pobierz dane z redshift, pobierz dane z MySQL, wykonaj kilka operacji na dwóch zestawach wyników, połącz je, a następnie prześlij wyniki do Amazon S3, wyślij e-mail.

Z tego, co przeczytałem online, można uruchamiać zlecenia przepływu powietrza, wykonując w wierszu poleceń airflow .... Zastanawiam się, czy istnieje python api, który może wykonać to samo.

Dzięki.

Answer 1

Pomoże ci tutaj Airflow REST API Plugin. Po wykonaniu instrukcji instalacji wtyczki wystarczy kliknąć następujący adres 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}, zastępując dag_id identyfikatorem dag, pomijając run_id lub podać unikalny identyfikator i przekazując kodowany json dla conf (z którykolwiek z parametrów, które potrzebujesz w wyzwalanym dag).

Oto przykład funkcji JavaScript, który używa jQuery do wywołania API Airflow:

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

Powinieneś spojrzeć na Airflow HTTP Sensor dla swoich potrzeb. Możesz użyć tego, aby wywołać dag.

Answer 3

Nowa opcja w strumieniu powietrza jest eksperymentalny, ale wbudowany, API punkt końcowy w nowsza kompilacje 1.7 i 1.8. Umożliwia to uruchomienie usługi REST na serwerze przepływów powietrza w celu odsłuchania portu i zaakceptowania zadań cli.

Sam mam tylko ograniczone doświadczenie, ale udało mi się uruchomić testowe dagi z sukcesem. Dla dokumentów:

/api/experimental/dags/<DAG_ID>/dag_runs tworzy dag_run dla danego ID dag (POST).

To zaplanuje natychmiastowe uruchomienie dowolnego dag, który chcesz uruchomić. Nadal jednak korzysta z harmonogramu, czekając, aż bicie serca zobaczy, że dag działa i przekazuje zadania pracownikowi. Jest to dokładnie to samo zachowanie co CLI, więc nadal uważam, że pasuje do twojego przypadku użycia.

Dokumentacja jak go skonfigurować jest dostępna tutaj: https://airflow.apache.org/api.html

Istnieje kilka prostych przykład klienci w github też pod przepływu powietrza/API/klientów