Fan-out/fan-in patroon

• Het quickstart-artikel voltooien
• Kloneer of download het sampleproject van GitHub
  • Geïsoleerde modelmonsters
  • Voorbeelden van in-process modellen

• Het quickstart-artikel voltooien
• Clone of download het voorbeeldproject van GitHub

• Het quickstart-artikel voltooien
• Clone of download het voorbeeldproject van GitHub

• Het quickstart-artikel voltooien

De volgende code laat zien hoe u de orchestratorfunctie implementeert:

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;
}

De functie maakt gebruik van de standaard function.json voor orchestratorfuncties.

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

De volgende code laat zien hoe u de orchestratorfunctie implementeert:

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)

Let op de yield context.task_all(tasks); regel. De code levert de afzonderlijke aanroepen niet op E2_CopyFileToBlob, zodat ze parallel worden uitgevoerd. Wanneer de orchestrator de takenlijst naar context.task_all doorgeeft, wordt er een taak geretourneerd die pas wordt voltooid als alle kopieerbewerkingen zijn afgerond. Als u bekend bent met asyncio.gather Python, is dit concept niet nieuw voor u. Met de Durable Functions-extensie worden deze taken gelijktijdig uitgevoerd op meerdere virtuele machines en is de end-to-end uitvoering bestand tegen het herstarten van processen.

Nadat de orchestrator context.task_all yieldt, worden alle functieaanroepen voltooid en retourneren ze waarden. Elke aanroep van E2_CopyFileToBlob geeft het aantal geüploade bytes terug, dus je kunt het totale aantal bytes berekenen door alle retourwaarden bij elkaar op te tellen.

De orchestrator krijgt een lijst met bestanden en splitst de taak vervolgens op om elk bestand parallel naar blobopslag te kopiëren met behulp van -NoWait. Nadat alle parallelle taken zijn voltooid, worden de resultaten opgeteld.

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;
}

De afzonderlijke aanroepen van E2_CopyFileToBlob worden niet elk afzonderlijk afgewacht, dus worden ze parallel uitgevoerd. Wanneer de orchestrator de takenlijst doorgeeft aan ctx.allOf(parallelTasks), geeft hij een taak terug die pas voltooid is als alle kopieerbewerkingen zijn voltooid. Nadat alle taken zijn voltooid, telt de orchestrator de resultaten op om het totale aantal bytes te uploaden.

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;
    }
}

Gebruik Task.WhenAll() dit om te wachten tot alle parallelle taken zijn voltooid. De Durable Task SDK zorgt ervoor dat de taken gelijktijdig op meerdere computers kunnen worden uitgevoerd en dat de uitvoering bestand is tegen het opnieuw opstarten van processen.

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;
};

Gebruik whenAll() dit om te wachten tot alle parallelle taken zijn voltooid. De Durable Task SDK zorgt ervoor dat de taken gelijktijdig op meerdere computers kunnen worden uitgevoerd en dat de uitvoering bestand is tegen het opnieuw opstarten van processen.

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

Gebruik task.when_all() dit om te wachten tot alle parallelle taken zijn voltooid. De Durable Task SDK zorgt ervoor dat de taken gelijktijdig op meerdere computers kunnen worden uitgevoerd en dat de uitvoering bestand is tegen het opnieuw opstarten van processen.

Dit voorbeeld is beschikbaar voor .NET, JavaScript, Java en Python.

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();

Gebruik ctx.allOf(tasks).await() dit om te wachten tot alle parallelle taken zijn voltooid. De Durable Task SDK zorgt ervoor dat de taken gelijktijdig op meerdere computers kunnen worden uitgevoerd en dat de uitvoering bestand is tegen het opnieuw opstarten van processen.

Het bestandfunction.jsonE2_GetFileList ziet eruit zoals in het volgende voorbeeld:

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

Dit is de implementatie:

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

De E2_GetFileList activiteit verzamelt recursief bestandspaden uit de opgegeven map:

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());
            }
        }
    }
}

Het function.json-bestand voor E2_CopyFileToBlob is vergelijkbaar eenvoudig:

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

De Python-implementatie maakt gebruik van de Azure Storage SDK voor Python om de bestanden te uploaden naar Azure Blob Storage.

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

De E2_CopyFileToBlob activiteit leest een bestand en uploadt het naar Azure Blob Storage:

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;
};

In tegenstelling tot orchestrators kunnen activiteiten I/O-bewerkingen uitvoeren, zoals HTTP-aanroepen, databasequery's en bestandstoegang.

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}

Dit voorbeeld wordt weergegeven voor .NET, JavaScript, Java en Python.

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,
  };
};

In tegenstelling tot orchestrators kunnen activiteiten I/O-bewerkingen uitvoeren, zoals HTTP-aanroepen, databasequery's en bestandstoegang.

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
    }

Dit voorbeeld wordt weergegeven voor .NET, JavaScript, Java en Python.

In het Java voorbeeld voegt de orkestrator resultaten samen nadat ctx.allOf(tasks).await() teruggeeft.

// 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}`);

Maak de DurableTaskAzureManagedClientBuilder met behulp van een verbindingsreeks naar de Durable Task Scheduler. Gebruik scheduleNewOrchestration om een orkestratie te starten en waitForOrchestrationCompletion om te wachten op voltooiing.

# 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}")

Dit voorbeeld wordt weergegeven voor .NET, JavaScript, Java en Python.

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));