Kann ich Luigi mit Python Sellerie

Question

verwenden Ich verwende Sellerie für meine Webanwendung. Sellerie führt Aufgaben Eltern, die dann weiter pipline von Aufgaben

Die Probleme mit Sellerie führt

1. ich nicht Abhängigkeitsdiagramm und Visualizer i mit luigi bekommen bekommen, um zu sehen, was ist der Status meiner Eltern-Task

2. Sellery bietet keinen Mechanismus, um die fehlerhafte Pipeline neu zu starten und von dort aus zu starten, wo sie fehlgeschlagen ist.

Diese zwei Dinge kann ich leicht von Luigi bekommen.

So dachte ich, dass sobald Sellerie die Elternaufgabe ausführt, dann führe ich innerhalb dieser Aufgabe die Luigi Piplein.

Gibt es irgendein Problem damit, d. H. Ich muss die Sellerie Arbeiter auf der Grundlage der Warteschlangengröße automatisch skalieren. Wird sich das auf Luigi-Arbeiter über mehrere Maschinen auswirken?

Answer 1

nie versucht, aber ich denke, dass es möglich sein soll, eine luigi Aufgabe Form in einer Sellerie Aufgabe zu nennen, genauso wie Sie es von Python-Code im Allgemeinen tun:

from foobar import MyTask 
from luigi import scheduler 

task = MyTask(123, 'another parameter value') 
sch = scheduler.CentralPlannerScheduler() 
w = worker.Worker(scheduler=sch) 
w.add(task) 
w.run() 

über die Warteschlange und Sellerie Arbeiter Skalierung Wenn Sie zu viele Sellerie-Arbeiter haben, die Luigi-Aufgaben aufrufen, müssen Sie natürlich Ihren luigi-Scheduler/-Daemon so skalieren, dass er die Anzahl der API-Anfragen bewältigen kann (jedes Mal, wenn Sie eine Aufgabe aufrufen, stoßen Sie auf die Luigi-Scheduler-API) , alle N Sekunden - es hängt von Ihrer Konfiguration ab - Ihre Aufgaben werden die Scheduler-API treffen, um zu sagen "Ich bin am Leben", jedes Mal, wenn eine Aufgabe mit Fehler oder Erfolg beendet wurde, Sie die Scheduler API getroffen haben und so weiter.

Also schauen Sie sich Ihren Scheduler genau an, um zu sehen, ob er zu viele HTTP-Anfragen bekommt oder ob seine Datenbank ein Flaschenhals ist (Luigi verwendet standardmäßig einen SQLite, aber Sie können ihn leicht zu mysql o postgres ändern) .

UPDATE:

Seit version 2.7.0 hat luigi.scheduler.CentralPlannerScheduler-luigi.scheduler.Scheduler als you may see here so der obige Code sollte jetzt umbenannt:

from foobar import MyTask 
from luigi import scheduler 

task = MyTask(123, 'another parameter value') 
sch = scheduler.Scheduler() 
w = worker.Worker(scheduler=sch) 
w.add(task) 
w.run()