Muster Auffächern nach außen/innen

• Schließen Sie den Schnellstartartikel ab.
• Klone oder lade das Sample-Projekt von GitHub herunter
  • Isolierte Modellproben
  • Beispiele für In-Processing-Modelle

• Schließen Sie den Schnellstartartikel ab.
• Clonen oder laden Sie das Beispielprojekt von GitHub herunter

• Schließen Sie den Schnellstartartikel ab.
• Clonen oder laden Sie das Beispielprojekt von GitHub herunter

• Schließen Sie den Schnellstartartikel ab.

Der folgende Code veranschaulicht die Implementierung der Orchestratorfunktion:

using System.IO;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

namespace SampleApp;

public static class BackupSiteContent
{
    [Function("E2_BackupSiteContent")]
    public static async Task<long> Run(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        string rootDirectory = context.GetInput<string>()?.Trim();
        if (string.IsNullOrEmpty(rootDirectory))
        {
            rootDirectory = Directory.GetParent(typeof(BackupSiteContent).Assembly.Location)!.FullName;
        }

        string[] files = await context.CallActivityAsync<string[]>("E2_GetFileList", rootDirectory);

        Task<long>[] tasks = files
            .Select(file => context.CallActivityAsync<long>("E2_CopyFileToBlob", file))
            .ToArray();

        long[] results = await Task.WhenAll(tasks);
        return results.Sum();
    }
}

[FunctionName("E2_BackupSiteContent")]
public static async Task<long> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext backupContext)
{
    string rootDirectory = backupContext.GetInput<string>()?.Trim();
    if (string.IsNullOrEmpty(rootDirectory))
    {
        rootDirectory = Directory.GetParent(typeof(BackupSiteContent).Assembly.Location).FullName;
    }

    string[] files = await backupContext.CallActivityAsync<string[]>(
        "E2_GetFileList",
        rootDirectory);

    var tasks = new Task<long>[files.Length];
    for (int i = 0; i < files.Length; i++)
    {
        tasks[i] = backupContext.CallActivityAsync<long>(
            "E2_CopyFileToBlob",
            files[i]);
    }

    await Task.WhenAll(tasks);

    long totalBytes = tasks.Sum(t => t.Result);
    return totalBytes;
}

Die Funktion verwendet die Standarddatei function.json für Orchestratorfunktionen.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

Der folgende Code veranschaulicht die Implementierung der Orchestratorfunktion:

import azure.functions as func
import azure.durable_functions as df


def orchestrator_function(context: df.DurableOrchestrationContext):

    root_directory: str = context.get_input()

    if not root_directory:
        raise Exception("A directory path is required as input")

    files = yield context.call_activity("E2_GetFileList", root_directory)
    tasks = []
    for file in files:
        tasks.append(context.call_activity("E2_CopyFileToBlob", file))
    
    results = yield context.task_all(tasks)
    total_bytes = sum(results)
    return total_bytes

main = df.Orchestrator.create(orchestrator_function)

Beachten Sie die Zeile yield context.task_all(tasks);. Der Code liefert nicht die einzelnen Aufrufe von E2_CopyFileToBlob, daher werden sie parallel ausgeführt. Wenn der Orchestrator das Aufgabenarray an context.task_all übergibt, wird eine Aufgabe zurückgegeben, die erst abgeschlossen wird, wenn alle Kopiervorgänge abgeschlossen sind. Wenn Sie mit asyncio.gather Python vertraut sind, ist dieses Konzept ihnen nicht neu. Mit der Erweiterung "Durable Functions" werden diese Aufgaben gleichzeitig auf mehreren virtuellen Computern ausgeführt, und die End-to-End-Ausführung ist stabil für die Prozesswiederverwendung.

Wenn der Orchestrator context.task_all liefert, sind alle Funktionsaufrufe abgeschlossen und geben Werte zurück. Jeder Aufruf gibt die E2_CopyFileToBlob Anzahl der hochgeladenen Bytes zurück, sodass man die Summe der Gesamtbytes berechnen kann, indem man alle Rückgabewerte zusammenzählt.

Der Orchestrator erhält eine Liste von Dateien und fächert dann auf, um jede Datei mithilfe von -NoWait parallel in den Blob-Speicher zu kopieren. Nachdem alle parallelen Aufgaben abgeschlossen sind, werden die Ergebnisse zusammengezählt.

param($Context)

$rootDirectory = $Context.Input

# Get all files in the directory
$files = Invoke-DurableActivity -FunctionName 'E2_GetFileList' -Input $rootDirectory

# Fan-out: schedule parallel uploads for each file
$parallelTasks = @()
foreach ($file in $files) {
    $parallelTasks += Invoke-DurableActivity -FunctionName 'E2_CopyFileToBlob' -Input $file -NoWait
}

# Fan-in: wait for all uploads and sum the results
$results = Wait-DurableTask -Task $parallelTasks
$totalBytes = ($results | Measure-Object -Sum).Sum

$totalBytes

@FunctionName("E2_BackupSiteContent")
public long backupSiteContent(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {

    String rootDirectory = ctx.getInput(String.class);

    // Get all files in the directory
    List<String> files = ctx.callActivity("E2_GetFileList", rootDirectory, List.class).await();

    // Fan-out: schedule parallel uploads for each file
    List<Task<Long>> parallelTasks = new ArrayList<>();
    for (String file : files) {
        parallelTasks.add(ctx.callActivity("E2_CopyFileToBlob", file, Long.class));
    }

    // Fan-in: wait for all uploads and sum the results
    List<Long> results = ctx.allOf(parallelTasks).await();
    long totalBytes = 0;
    for (Long bytes : results) {
        totalBytes += bytes;
    }

    return totalBytes;
}

Die einzelnen Aufrufe zu E2_CopyFileToBlob werden nicht einzeln abgewartet, daher laufen sie parallel. Wenn der Orchestrator die Aufgabenliste an ctx.allOf(parallelTasks)übergibt, gibt er eine Aufgabe zurück, die erst abgeschlossen ist, wenn alle Kopieroperationen abgeschlossen sind. Nachdem alle Aufgaben abgeschlossen sind, summiert der Orchestrator die Ergebnisse, um die Gesamtzahl der Bytes hochzuladen.

using Microsoft.DurableTask;
using System.Collections.Generic;
using System.Threading.Tasks;

[DurableTask]
public class ParallelProcessingOrchestration : TaskOrchestrator<List<string>, Dictionary<string, int>>
{
    public override async Task<Dictionary<string, int>> RunAsync(
        TaskOrchestrationContext context, List<string> workItems)
    {
        // Step 1: Fan-out by creating a task for each work item in parallel
        var processingTasks = new List<Task<Dictionary<string, int>>>();

        foreach (string workItem in workItems)
        {
            // Create a task for each work item (fan-out)
            Task<Dictionary<string, int>> task = context.CallActivityAsync<Dictionary<string, int>>(
                nameof(ProcessWorkItemActivity), workItem);
            processingTasks.Add(task);
        }

        // Step 2: Wait for all parallel tasks to complete
        Dictionary<string, int>[] results = await Task.WhenAll(processingTasks);

        // Step 3: Fan-in by aggregating all results
        Dictionary<string, int> aggregatedResults = await context.CallActivityAsync<Dictionary<string, int>>(
            nameof(AggregateResultsActivity), results);

        return aggregatedResults;
    }
}

Verwenden Sie Task.WhenAll(), um auf den Abschluss aller parallelen Aufgaben zu warten. Das Durable Task SDK stellt sicher, dass die Aufgaben gleichzeitig auf mehreren Computern ausgeführt werden können, und die Ausführung ist robust für Neustarts.

import {
  OrchestrationContext,
  TOrchestrator,
  whenAll,
} from "@microsoft/durabletask-js";

const fanOutFanInOrchestrator: TOrchestrator = async function* (
  ctx: OrchestrationContext,
  workItems: string[]
): any {
  // Fan-out: create a task for each work item in parallel
  const tasks = workItems.map((item) => ctx.callActivity(processWorkItem, item));

  // Wait for all parallel tasks to complete
  const results: number[] = yield whenAll(tasks);

  // Fan-in: aggregate all results
  const aggregatedResult = yield ctx.callActivity(aggregateResults, results);

  return aggregatedResult;
};

Verwenden Sie whenAll(), um auf den Abschluss aller parallelen Aufgaben zu warten. Das Durable Task SDK stellt sicher, dass die Aufgaben gleichzeitig auf mehreren Computern ausgeführt werden können, und die Ausführung ist robust für Neustarts.

from durabletask import task

def fan_out_fan_in_orchestrator(ctx: task.OrchestrationContext, work_items: list) -> dict:
    """Orchestrator demonstrating fan-out/fan-in pattern."""

    # Fan-out: Create a task for each work item
    parallel_tasks = []
    for item in work_items:
        parallel_tasks.append(ctx.call_activity(process_work_item, input=item))

    # Wait for all tasks to complete
    results = yield task.when_all(parallel_tasks)

    # Fan-in: Aggregate all the results
    final_result = yield ctx.call_activity(aggregate_results, input=results)

    return final_result

Verwenden Sie task.when_all(), um auf den Abschluss aller parallelen Aufgaben zu warten. Das Durable Task SDK stellt sicher, dass die Aufgaben gleichzeitig auf mehreren Computern ausgeführt werden können, und die Ausführung ist robust für Neustarts.

Dieses Beispiel ist für .NET, JavaScript, Java und Python verfügbar.

import com.microsoft.durabletask.*;
import java.util.List;
import java.util.stream.Collectors;

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "FanOutFanIn_WordCount"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                List<?> inputs = ctx.getInput(List.class);

                // Fan-out: Create a task for each input item
                List<Task<Integer>> tasks = inputs.stream()
                    .map(input -> ctx.callActivity("CountWords", input.toString(), Integer.class))
                    .collect(Collectors.toList());

                // Wait for all parallel tasks to complete
                List<Integer> allResults = ctx.allOf(tasks).await();

                // Fan-in: Aggregate results
                int totalCount = allResults.stream().mapToInt(Integer::intValue).sum();

                ctx.complete(totalCount);
            };
        }
    })
    .build();

Verwenden Sie ctx.allOf(tasks).await(), um auf den Abschluss aller parallelen Aufgaben zu warten. Das Durable Task SDK stellt sicher, dass die Aufgaben gleichzeitig auf mehreren Computern ausgeführt werden können, und die Ausführung ist robust für Neustarts.

Die function.json Datei E2_GetFileList sieht wie im folgenden Beispiel aus:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "rootDirectory",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Dies ist die Implementierung:

import os
from os.path import dirname
from typing import List

def main(rootDirectory: str) -> List[str]:

    all_file_paths = []
    # We walk the file system
    for path, _, files in os.walk(rootDirectory):
        # We copy the code for activities and orchestrators
        if "E2_" in path:
            # For each file, we add their full-path to the list
            for name in files:
                if name == "__init__.py" or name == "function.json":
                    file_path = os.path.join(path, name)
                    all_file_paths.append(file_path)
    
    return all_file_paths

Die Aktivität E2_GetFileList sammelt rekursiv Dateipfade aus dem angegebenen Verzeichnis:

param($rootDirectory)

Get-ChildItem -Path $rootDirectory -Recurse -File | Select-Object -ExpandProperty FullName

@FunctionName("E2_GetFileList")
public List<String> getFileList(
        @DurableActivityTrigger(name = "rootDirectory") String rootDirectory) {

    File root = new File(rootDirectory);
    List<String> files = new ArrayList<>();
    collectFiles(root, files);
    return files;
}

private void collectFiles(File directory, List<String> files) {
    File[] entries = directory.listFiles();
    if (entries != null) {
        for (File entry : entries) {
            if (entry.isDirectory()) {
                collectFiles(entry, files);
            } else {
                files.add(entry.getAbsolutePath());
            }
        }
    }
}

Die Datei function.json für E2_CopyFileToBlob ist ebenso einfach:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "filePath",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Die Python-Implementierung verwendet das Azure Storage SDK für Python, um die Dateien in Azure Blob Storage hochzuladen.

import os
import pathlib
from azure.storage.blob import BlobServiceClient
from azure.core.exceptions import ResourceExistsError

connect_str = os.getenv('AzureWebJobsStorage')

def main(filePath: str) -> str:
    # Create the BlobServiceClient object which will be used to create a container client
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)
    
    # Create a unique name for the container
    container_name = "backups"
    
    # Create the container if it does not exist
    try:
        blob_service_client.create_container(container_name)
    except ResourceExistsError:
        pass

    # Create a blob client using the local file name as the name for the blob
    parent_dir, fname = pathlib.Path(filePath).parts[-2:] # Get last two path components
    blob_name = parent_dir + "_" + fname
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)

    # Count bytes in file
    byte_count = os.path.getsize(filePath)

    # Upload the created file
    with open(filePath, "rb") as data:
        blob_client.upload_blob(data)

    return byte_count

Die E2_CopyFileToBlob Aktivität liest eine Datei und lädt sie in Azure Blob Storage hoch:

param($filePath)

$storageContext = New-AzStorageContext -ConnectionString $env:AzureWebJobsStorage
$container = "backups"

# Create the container if it doesn't exist
New-AzStorageContainer -Name $container -Context $storageContext -ErrorAction SilentlyContinue

$blobName = $filePath.Substring([System.IO.Path]::GetPathRoot($filePath).Length).Replace('\', '/')
Set-AzStorageBlobContent -File $filePath -Container $container -Blob $blobName -Context $storageContext -Force

(Get-Item $filePath).Length

@FunctionName("E2_CopyFileToBlob")
public long copyFileToBlob(
        @DurableActivityTrigger(name = "filePath") String filePath) throws Exception {

    File file = new File(filePath);
    long byteCount = file.length();

    String blobPath = filePath
            .substring(File.listRoots()[0].getPath().length())
            .replace('\\', '/');

    String connectionString = System.getenv("AzureWebJobsStorage");
    BlobContainerClient containerClient = new BlobContainerClientBuilder()
            .connectionString(connectionString)
            .containerName("backups")
            .buildClient();
    containerClient.createIfNotExists();

    BlobClient blobClient = containerClient.getBlobClient(blobPath);
    blobClient.uploadFromFile(filePath, true);

    return byteCount;
}

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;
using System.Threading.Tasks;

[DurableTask]
public class ProcessWorkItemActivity : TaskActivity<string, Dictionary<string, int>>
{
    private readonly ILogger<ProcessWorkItemActivity> _logger;

    public ProcessWorkItemActivity(ILogger<ProcessWorkItemActivity> logger)
    {
        _logger = logger;
    }

    public override Task<Dictionary<string, int>> RunAsync(TaskActivityContext context, string workItem)
    {
        _logger.LogInformation("Processing work item: {WorkItem}", workItem);

        // Process the work item (where you do the actual work)
        var result = new Dictionary<string, int>
        {
            { workItem, workItem.Length }
        };

        return Task.FromResult(result);
    }
}

import { ActivityContext } from "@microsoft/durabletask-js";

const processWorkItem = async (
  _ctx: ActivityContext,
  item: string
): Promise<number> => {
  console.log(`Processing work item: "${item}"`);
  return item.length;
};

Im Gegensatz zu Orchestratoren können Aktivitäten E/A-Vorgänge wie HTTP-Aufrufe, Datenbankabfragen und Dateizugriff ausführen.

from durabletask import task

def process_work_item(ctx: task.ActivityContext, item: int) -> dict:
    """Activity processing a single work item."""
    # Process the work item (where you do the actual work)
    result = item * item
    return {"item": item, "result": result}

Dieses Beispiel wird für .NET, JavaScript, Java und Python gezeigt.

import java.util.StringTokenizer;

// Activity registration
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "CountWords"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            StringTokenizer tokenizer = new StringTokenizer(input);
            return tokenizer.countTokens();
        };
    }
})

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;
using System.Threading.Tasks;

[DurableTask]
public class AggregateResultsActivity : TaskActivity<Dictionary<string, int>[], Dictionary<string, int>>
{
    private readonly ILogger<AggregateResultsActivity> _logger;

    public AggregateResultsActivity(ILogger<AggregateResultsActivity> logger)
    {
        _logger = logger;
    }

    public override Task<Dictionary<string, int>> RunAsync(
        TaskActivityContext context, Dictionary<string, int>[] results)
    {
        _logger.LogInformation("Aggregating {Count} results", results.Length);

        // Combine all results into one aggregated result
        var aggregatedResult = new Dictionary<string, int>();

        foreach (var result in results)
        {
            foreach (var kvp in result)
            {
                aggregatedResult[kvp.Key] = kvp.Value;
            }
        }

        return Task.FromResult(aggregatedResult);
    }
}

import { ActivityContext } from "@microsoft/durabletask-js";

const aggregateResults = async (
  _ctx: ActivityContext,
  results: number[]
): Promise<object> => {
  const total = results.reduce((sum, val) => sum + val, 0);
  return {
    totalItems: results.length,
    sum: total,
    average: results.length > 0 ? total / results.length : 0,
  };
};

Im Gegensatz zu Orchestratoren können Aktivitäten E/A-Vorgänge wie HTTP-Aufrufe, Datenbankabfragen und Dateizugriff ausführen.

from durabletask import task

def aggregate_results(ctx: task.ActivityContext, results: list) -> dict:
    """Activity aggregating results from multiple work items."""
    sum_result = sum(item["result"] for item in results)
    return {
        "total_items": len(results),
        "sum": sum_result,
        "average": sum_result / len(results) if results else 0
    }

Dieses Beispiel wird für .NET, JavaScript, Java und Python gezeigt.

Im Java Beispiel aggregiert der Orchestrator Ergebnisse, nachdem ctx.allOf(tasks).await() zurückgegeben wurde.

// Schedule the orchestration with a list of work items
var workItems = new List<string> { "item1", "item2", "item3", "item4", "item5" };
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    nameof(ParallelProcessingOrchestration), workItems);

// Wait for completion
var result = await client.WaitForInstanceCompletionAsync(instanceId, getInputsAndOutputs: true);
Console.WriteLine($"Result: {result.ReadOutputAs<Dictionary<string, int>>().Count} items processed");

import {
  DurableTaskAzureManagedClientBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

const connectionString =
  process.env.DURABLE_TASK_SCHEDULER_CONNECTION_STRING ||
  "Endpoint=http://localhost:8080;Authentication=None;TaskHub=default";

const client = new DurableTaskAzureManagedClientBuilder()
  .connectionString(connectionString)
  .build();

const workItems = ["item1", "item2", "item3", "item4", "item5"];
const instanceId = await client.scheduleNewOrchestration(fanOutFanInOrchestrator, workItems);
const state = await client.waitForOrchestrationCompletion(instanceId, true, 30);
console.log(`Result: ${state?.serializedOutput}`);

Erstellen Sie den DurableTaskAzureManagedClientBuilder mithilfe einer Verbindungszeichenfolge für den Durable Task Scheduler. Verwenden Sie scheduleNewOrchestration zum Starten einer Orchestrierung und waitForOrchestrationCompletion zum Warten auf den Abschluss.

# Schedule the orchestration with a list of work items
work_items = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
instance_id = client.schedule_new_orchestration(fan_out_fan_in_orchestrator, input=work_items)

# Wait for completion
result = client.wait_for_orchestration_completion(instance_id, timeout=60)
print(f"Result: {result.serialized_output}")

Dieses Beispiel wird für .NET, JavaScript, Java und Python gezeigt.

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

// Schedule the orchestration with a list of strings
List<String> sentences = Arrays.asList(
    "Hello, world!",
    "The quick brown fox jumps over the lazy dog.",
    "Always remember you are absolutely unique.");

String instanceId = client.scheduleNewOrchestrationInstance(
    "FanOutFanIn_WordCount",
    new NewOrchestrationInstanceOptions().setInput(sentences));

// Wait for completion
OrchestrationMetadata result = client.waitForInstanceCompletion(instanceId, Duration.ofSeconds(30), true);
System.out.println("Total word count: " + result.readOutputAs(int.class));