Arbeitsitem-Filterung im Durable Task Scheduler

Die Filterung von Arbeitselementen ist eine Routingfunktion im Durable Task Scheduler, die jedes Arbeitselement nur an die Worker weiterleitet, die deklarieren, dass sie die entsprechende Orchestrierung oder Aktivität hosten. Produktionsworkflows kombinieren oft Schritte mit unterschiedlichen Rechenbedarfs, wie leichte Koordination, großflächige Aktivitäten und spezialisierte Verarbeitung. Ohne eine Möglichkeit, jeden Schritt unabhängig zu routen, konkurrieren diese Schritte um die gleiche Kapazität oder zwingen dich, für jeden eine separate Koordinationsinfrastruktur aufzubauen.

Dieses Routing basiert auf einer flachen Netzwerktopologie : Jeder Worker verbindet sich Outbound mit dem Scheduler und nicht direkt mit anderen Workern. Daher koordiniert der Scheduler das Routing für Sie, sodass keine Diensterkennung, keine eingehenden Ports und keine dedizierte Kommunikationsschicht erforderlich sind, die Sie selbst entwickeln und betreiben müssen.

In diesem Artikel erfahren Sie, wann Sie Arbeitsitem-Filterung verwenden sollten und wie der Scheduler jeden Schritt zur passenden Berechnung routet.

Wann sollte man Arbeitsitem-Filterung verwenden

Arbeitsitem-Filterung ist am nützlichsten, wenn die Schritte eines einzelnen Workflows unterschiedliche Laufzeitanforderungen haben. Ohne sie läuft jeder Schritt im selben Rechenpool, sodass man den gesamten Workflow für den anspruchsvollsten Schritt dimensioniert, und spezialisierte oder immer-eingeschaltete Komponenten konkurrieren mit allem anderen um die Kapazität. Die folgenden Szenarien zeigen, wie das Routing jedes Schritts an die dafür passende Rechenressource dieses Problem löst.

Kombinieren Sie stets verfügbare und Scale-to-Zero-Rechenressourcen

Einige Teile einer Anwendung laufen kontinuierlich, während andere im Leerlauf nichts kosten sollten. Durch die Filterung von Arbeitselementen können Sie jeden Schritt auf der Rechenebene platzieren, die zu seiner Arbeitslast passt.

• Führe Always-on-Komponenten auf Ready Compute aus, wie zum Beispiel Azure Kubernetes Service (AKS), damit Orchestrierungen sofort beginnen.
• Führen Sie sprunghaft anfallende, ereignisgesteuerte Komponenten auf serverlosen Computediensten wie Azure Container Apps oder Azure Functions aus, die bei eingehenden Workloads von null hochskalieren und bei nachlassender Last wieder auf null herunterskalieren.

Das Filtern stellt sicher, dass jede Ebene nur die Aufgaben abruft, die sie hostet, sodass Komponenten mit Lastspitzen dauerhaft aktive Komponenten nie verdrängen und auf null skalierte Computeressourcen nur für die Aktivitäten hochgefahren werden, für die sie zuständig sind. Für Details zur Autoskalierung siehe Configure Autoscale for Azure Container Apps.

Berechnung der richtigen Größe pro Aktivitätsschritt

Wenn ein Schritt viel mehr Speicher oder CPU verbraucht als der andere, verschwendet die Dimensionierung der gesamten Bereitstellung für diesen Schritt Ressourcen. Work Item-Filterung ermöglicht es dir, Ressourcen pro Schritt innerhalb eines einzigen Workflows zuzuweisen.

• Geben Sie der ressourcenintensiven Aktivität – zum Beispiel einem Datenaggregationsschritt, der einen großen Datensatz in den Speicher lädt – einen eigenen Worker auf einer größeren SKU, und registrieren Sie dort nur diese Aktivität.
• Führe jeden zweiten Schritt bei kleineren, günstigeren Arbeitern durch.

Man dimensioniert jeden Mitarbeiter entsprechend der Arbeit, die er leistet, anstatt eine einheitliche Bereitstellung zu skalieren, um die Anforderungen des größten Schrittes zu erfüllen.

Routenarbeit zu spezialisierter Hardware

Ein einzelner Schritt in einem Workflow kann spezielle Hardware erfordern, die der Rest der Anwendung nicht benötigt. Zum Beispiel läuft ein Workflow hauptsächlich in Azure Functions, aber ein Schritt erfordert eine GPU, um ein KI-Modell auszuführen. Der Scheduler übermittelt diesen Schritt an einen Worker, der in Azure Container Apps auf GPU-fähiger Infrastruktur läuft, und der Workflow läuft dann in Azure Functions weiter.

Setzen Sie einen GPU-unterstützten Worker ein, der nur die GPU-Aktivität registriert, und lassen Sie den Rest des Workflows auf Ihrer Standard-Berechnung ein. Der Scheduler routet die Arbeitselemente jeder Aktivität an den dafür entwickelten Worker, sodass die GPU-Infrastruktur nur die Arbeit ausführt, die sie benötigt.

Wie das Filtern von Arbeitsgegenständen funktioniert

Der Durable Task Scheduler unterteilt die Orchestrierungsausführung in diskrete Arbeitseinheiten, die Arbeitselemente genannt werden. Es gibt drei Haupttypen von Arbeitsgegenständen:

• Orchestrierungs-Arbeitselemente: Führen Sie Orchestrator-Code aus, der den Workflow koordiniert.
• Aktivitätsaufgaben: Führen Sie einzelne Aktivitätsfunktionen aus, die die eigentliche Arbeit ausführen, wie zum Beispiel das Validieren einer Bestellung oder den Versand.
• Arbeitselemente der Entität: Führen Sie Operationen auf dauerhaften Entitäten aus, die der Scheduler über den Namen der registrierten Entität routet.

Ein Worker ist jeder Prozess, der mit dem Scheduler verbunden ist, um Arbeitselemente zu ziehen und auszuführen. Ein Mitarbeiter kann Orchestrierungen, Aktivitäten, Entitäten und jede Kombination davon ausrichten.

Jede Arbeitsaufgabe an einen verbundenen Mitarbeiter zu liefern, funktioniert gut, wenn ein einzelner Mitarbeiter oder eine Gruppe identischer Mitarbeiter jede Aufgabe im Workflow registriert. Das Filtern von Arbeitsgegenständen bietet dir eine Alternative: Du kannst bestimmte Arbeitsgegenstände an bestimmte Arbeiter weiterleiten. Jeder Worker registriert nur die Aufgaben, die er hostet, und der Scheduler leitet nur zugleiche Arbeitselemente an ihn weiter.

Wenn ein Worker startet, registriert er seine lokalen Aufgaben und aktiviert die Filterung. Das SDK erstellt eine Filterliste aus den registrierten Aufgaben und sendet sie an den Scheduler, wenn der Worker sich verbindet. Der Scheduler erstellt Warteschlangen pro Filter und routet jedes Arbeitselement an die Warteschlange, die zum Aufgabentyp passt. Ein Worker zieht nur aus Warteschlangen, die seinen Filtern entsprechen.

Zum Beispiel in einem Auftragsverarbeitungs-Workflow:

1. Der Orchestrator-Worker registriert OrderProcessingOrchestration und empfängt nur Orchestrierungsarbeitselemente.
2. Der Validator-Worker registriert ValidateOrder und erhält nur ValidateOrder Aktivitätsarbeitsobjekte.
3. Der Shipper-Worker registriert ShipOrder und erhält nur ShipOrder Aktivitäts-Arbeitselemente.

Das Routing ist streng. Es gibt keinen Rückfall auf einen anderen Mitarbeiter:

• Wenn kein Worker eine bestimmte Aufgabe registriert, bleiben die zugehörigen Work-Items Pending, bis sich ein passender Worker verbindet.
• Die Orchestrierung, die von dieser Aufgabe abhängig ist, verbleibt im Status Running, bis das ausstehende Arbeitselement abgeschlossen ist.

Diese Strenge stellt sicher, dass der Planerer niemals einen Arbeitspunkt an einen Mitarbeiter weitergibt, der ihn nicht bearbeiten kann.

Unterstützung für das Filtern von Arbeitsobjekten durch das Durable Task SDK

Aktiviere die Filterung von Arbeitsobjekten

Jeder Worker registriert nur die Aufgaben, die er hostet, und das SDK nutzt diese Registrierung, um die Workitem-Filter des Workers zu generieren. Die Aktivierung der Arbeitselement-Filterung unterscheidet sich leicht zwischen SDKs.

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddAllGeneratedTasks();
    })
    .UseDurableTaskScheduler(connectionString);

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions
        .createWorkerBuilder(connectionString)
        .addActivity(new TaskActivityFactory() {
            // ...
        })
        .useWorkItemFilters() // auto-generate from registered tasks
        .build();

WorkItemFilter filter = WorkItemFilter.newBuilder()
        .addOrchestration("OrderProcessingOrchestration")
        .addActivity("ValidateOrder")
        .build();

builder.useWorkItemFilters(filter);

worker.add_orchestrator(order_processing_orchestrator)
worker.add_activity(validate_order)
worker.use_work_item_filters()  # auto-generate from registered tasks

from durabletask.worker import WorkItemFilters, OrchestrationWorkItemFilter, ActivityWorkItemFilter

worker.use_work_item_filters(WorkItemFilters(
    orchestrations=[
        OrchestrationWorkItemFilter(name="order_processing_orchestrator"),
    ],
    activities=[
        ActivityWorkItemFilter(name="validate_order"),
    ],
))

package main

import (
	"github.com/microsoft/durabletask-go/api"
	"github.com/microsoft/durabletask-go/client"
	"github.com/microsoft/durabletask-go/durabletaskscheduler"
	"github.com/microsoft/durabletask-go/task"
)

func newFilteredWorker(options *durabletaskscheduler.Options, registry *task.TaskRegistry) (*client.TaskHubGrpcWorker, error) {
	return durabletaskscheduler.NewWorker(
		options, registry, api.DefaultLogger(), client.WithAutoWorkItemFilters())
}

Considerations

• Strenges Routing hat keine Ausweichmöglichkeit. Wenn kein angeschlossener Worker eine Aufgabe registriert, bleiben die Arbeitselemente der Aufgabe im Zustand „Ausstehend“, und die abhängige Orchestrierung verbleibt im Status Running. Stellen Sie sicher, dass mindestens ein Mitarbeiter jede Aufgabe hostet, die ein Workflow aufruft.
• Registrierungsinkonsistenzen bei Upgrades. Wenn Sie Aufgaben aufteilen oder umbenennen, setzen Sie die Arbeiter ein, die die neuen Aufgaben hosten, bevor Sie Orchestrierungen planen, die von ihnen abhängen.
• Überwachen Sie den ausstehenden Rückstand. Nutzen Sie das Überwachungs-Dashboard , um auf ausstehende Arbeitspunkte zu achten, die auf einen fehlenden oder ungesunden Mitarbeiter hinweisen.
• Automatische Skalierung. Wenn Sie Worker auf Azure Container Apps hosten, kann eine KEDA-Skalierungsregel (Kubernetes Event-driven Autoscaling) jeden Worker basierend auf dem Backlog der registrierten Arbeitselementtypen skalieren. Weitere Informationen finden Sie unter Autoscale konfigurieren für Azure Container Apps.

Nächste Schritte

• Identitätsbasierter Zugriff für den Durable Task Scheduler
• Konfigurieren Sie private Endpunkte für den Durable Task Scheduler
• Automatische Skalierung für Azure Container Apps konfigurieren