ファンアウト/ファンイン パターン

• クイックスタートの記事を完了する
• サンプルプロジェクトをGitHubからクローンまたはダウンロードしてください
  • 孤立したモデルサンプル
  • 進行中のモデルサンプル

• クイックスタートの記事を完了する
• GitHubからサンプルプロジェクトをクローンまたはダウンロードします

• クイックスタートの記事を完了する
• GitHubからサンプルプロジェクトをクローンまたはダウンロードします

• クイックスタートの記事を完了する

次のコードは、オーケストレーター関数の実装を示しています。

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

この関数では、オーケストレーター関数の標準的な function.json が使用されます。

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

次のコードは、オーケストレーター関数の実装を示しています。

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)

yield context.task_all(tasks); 行に注目してください。 このコードでは、 E2_CopyFileToBlobに対する個々の呼び出しは生成されないため、並列で実行されます。 オーケストレーターは、タスク配列を context.task_allに渡すと、すべてのコピー操作が完了するまで完了しないタスクを返します。 Python の asyncio.gather に慣れている場合、この概念は初めてではありません。 Durable Functions 拡張機能を使用すると、これらのタスクは複数の仮想マシンで同時に実行され、エンドツーエンドの実行はプロセスのリサイクルに対する回復性があります。

オーケストレーターが context.task_allを返した後、すべての関数呼び出しは完了し、値を返します。 各 E2_CopyFileToBlob 呼び出しはアップロードされたバイト数を返すため、すべての返還値を合計して合計バイト数を計算できます。

オーケストレーターはファイルの一覧を取得し、その後 -NoWait を使用して各ファイルを Blob Storage に並列でコピーします。 すべての並列作業が完了した後、結果が合計されます。

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

E2_CopyFileToBlobへの個別の電話は個別に待たされるわけではなく、並行して行われます。 オーケストレーターがタスクリストを ctx.allOf(parallelTasks)に渡すと、すべてのコピー操作が完了するまで完了しないタスクが返されます。 すべてのタスクが完了すると、オーケストレーターは結果を合計してアップロードされたバイト数を得ます。

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

Task.WhenAll()を使用して、すべての並列タスクが完了するまで待機します。 Durable Task SDK は、タスクを複数のマシンで同時に実行でき、実行が再起動を処理する回復性を確保します。

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

whenAll()を使用して、すべての並列タスクが完了するまで待機します。 Durable Task SDK は、タスクを複数のマシンで同時に実行でき、実行が再起動を処理する回復性を確保します。

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

task.when_all()を使用して、すべての並列タスクが完了するまで待機します。 Durable Task SDK は、タスクを複数のマシンで同時に実行でき、実行が再起動を処理する回復性を確保します。

このサンプルは、.NET、JavaScript、Java、および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();

ctx.allOf(tasks).await()を使用して、すべての並列タスクが完了するまで待機します。 Durable Task SDK は、タスクを複数のマシンで同時に実行でき、実行が再起動を処理する回復性を確保します。

ファイルは、次の例のようになります。

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

実装を次に示します。

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

E2_GetFileListアクティビティは指定されたディレクトリからファイルパスを再帰的に収集します:

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

function.json ファイルは E2_CopyFileToBlob と同様に単純です。

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

Python実装では、Python Azure Storage SDK を使用して、ファイルを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

E2_CopyFileToBlobアクティビティはファイルを読み取り、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;
};

オーケストレーターとは異なり、アクティビティは HTTP 呼び出し、データベース クエリ、ファイル アクセスなどの I/O 操作を実行できます。

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}

このサンプルは、.NET、JavaScript、Java、および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,
  };
};

オーケストレーターとは異なり、アクティビティは HTTP 呼び出し、データベース クエリ、ファイル アクセスなどの I/O 操作を実行できます。

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
    }

このサンプルは、.NET、JavaScript、Java、およびPythonについて示されています。

Javaサンプルでは、オーケストレーターは、ctx.allOf(tasks).await() が返された後に結果を集計します。

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

Durable Task Scheduler への接続文字列を使用して、DurableTaskAzureManagedClientBuilderを作成します。 scheduleNewOrchestrationを使用してオーケストレーションを開始し、waitForOrchestrationCompletionを使用して完了を待機します。

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

このサンプルは、.NET、JavaScript、Java、および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));