Cenário de fan-out/fan-in

Fan-out/fan-in executa várias funções em paralelo e agrega os resultados. Este artigo mostra um exemplo que usa Durable Functions para fazer backup de algum ou todo o conteúdo do site de um aplicativo para Azure Storage.

Pré-requisitos

C#

• Conclua o artigo de início rápido
• Clonar ou baixar o projeto de exemplos do GitHub (usa o modelo em processo)

JavaScript

Modelo de programação V3

• Conclua o artigo de início rápido
• Clone ou baixe o projeto de amostras do GitHub

Modelo de programação V4

• Conclua o artigo de início rápido
• Clone ou baixe o projeto de amostras do GitHub

Python

• Conclua o artigo de início rápido
• Clone ou baixe o projeto de amostras do GitHub

PowerShell

• Conclua o artigo de início rápido
• Clone ou baixe o projeto de amostras do GitHub

Java

• Conclua o artigo de início rápido

O fan-out/fan-in executa várias atividades em paralelo e agrega os resultados. Este artigo mostra como implementar o padrão usando os SDKs de Tarefa Durável para .NET, JavaScript, Python e Java.

Visão geral do cenário

Neste exemplo, as funções carregam todos os arquivos em um diretório especificado (recursivamente) para o armazenamento de blobs. Eles também contam o número total de bytes carregados.

Uma única função pode tratar tudo, mas ela não é dimensionada. Uma única execução de função é executada em uma VM (máquina virtual), portanto, a taxa de transferência é limitada a essa VM. A confiabilidade é outra preocupação. Se o processo falhar no meio do caminho ou levar mais de cinco minutos, o backup poderá terminar em um estado parcialmente concluído. Em seguida, reinicie o backup.

Uma abordagem mais robusta é usar duas funções separadas: uma enumera os arquivos e adiciona nomes de arquivo a uma fila e a outra lê da fila e carrega os arquivos no armazenamento de blobs. Essa abordagem melhora a taxa de transferência e a confiabilidade, mas você precisa configurar e gerenciar a fila. Mais importante, essa abordagem adiciona complexidade ao gerenciamento e coordenação de estado, como relatar o número total de bytes carregados.

Durable Functions fornece todos esses benefícios com pouca sobrecarga.

No exemplo a seguir, o orquestrador processa vários itens de trabalho em paralelo e, em seguida, agrega os resultados. Esse padrão é útil quando você precisa:

• Processar um lote de itens em que cada item pode ser processado independentemente
• Distribuir o trabalho em vários computadores para obter uma melhor taxa de transferência
• Agregar resultados de todas as operações paralelas

Sem o padrão fan-out/fan-in, você processa itens de forma sequencial, limitando a taxa de transferência, ou gerencia sua própria lógica de enfileiramento e coordenação, adicionando complexidade.

Os SDKs de Tarefa Durável lidam com paralelização e coordenação, portanto, o padrão é simples de implementar.

As funções

Este artigo descreve as funções no aplicativo de exemplo:

• E2_BackupSiteContent: uma função de orquestrador que chama E2_GetFileList para obter uma lista de arquivos para fazer backup e, em seguida, chama E2_CopyFileToBlob para cada arquivo.
• E2_GetFileList: uma função de atividade que retorna uma lista de arquivos em um diretório.
• E2_CopyFileToBlob: uma função de atividade que faz backup de um único arquivo para Azure Blob Storage.

Este artigo descreve os componentes no código de exemplo:

• ParallelProcessingOrchestration, fanOutFanInOrchestrator, fan_out_fan_in_orchestrator, ou FanOutFanIn_WordCount: um orquestrador que distribui o trabalho para várias atividades em paralelo, aguarda a conclusão de todas as atividades e então agrega os resultados.
• ProcessWorkItemActivity, processWorkItemou process_work_itemCountWords: uma atividade que processa um único item de trabalho.
• AggregateResultsActivity, aggregateResultsou aggregate_results: uma atividade que agrega resultados de todas as operações paralelas.

Orquestrador

Essa função de orquestrador faz o seguinte:

1. Usa rootDirectory como entrada.
2. Chama uma função para obter uma lista recursiva de arquivos em rootDirectory.
3. Faz chamadas de função paralelas para carregar cada arquivo para Azure Blob Storage.
4. Aguarda que todos os uploads sejam concluídos.
5. Retorna o número total de bytes carregados para Azure Blob Storage.

C#

Este é o código que implementa a função de orquestrador:

Modelo isolado

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

Observe a linha await Task.WhenAll(tasks);. O código não aguarda as chamadas individuais a E2_CopyFileToBlob, portanto, elas são executadas em paralelo. Quando o orquestrador passa a matriz de tarefas para Task.WhenAll, ele retorna uma tarefa que só é concluída quando todas as operações de cópia são concluídas. Se você estiver familiarizado com a TPL (Biblioteca Paralela de Tarefas) em .NET, esse padrão será familiar. A diferença é que essas tarefas podem estar em execução em várias máquinas virtuais simultaneamente e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Após o orquestrador aguardar Task.WhenAll, todas as chamadas de função são concluídas e retornam valores. Cada chamada para E2_CopyFileToBlob retorna o número de bytes enviados. Calcule o total adicionando os valores retornados.

Modelo em processo

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

Observação

O exemplo de modelo em processo utiliza pacotes obsoletos em processo. O código anterior mostra o modelo de trabalho isolado do .NET recomendado.

JavaScript

Modelo de programação V3

A função usa o function.json padrão para funções de orquestrador.

{
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

Este é o código que implementa a função de orquestrador:

const df = require("durable-functions");

module.exports = df.orchestrator(function* (context) {
    const rootDirectory = context.df.getInput();
    if (!rootDirectory) {
        throw new Error("A directory path is required as an input.");
    }

    const files = yield context.df.callActivity("E2_GetFileList", rootDirectory);

    // Backup Files and save Promises into array
    const tasks = [];
    for (const file of files) {
        tasks.push(context.df.callActivity("E2_CopyFileToBlob", file));
    }

    // wait for all the Backup Files Activities to complete, sum total bytes
    const results = yield context.df.Task.all(tasks);
    const totalBytes = results.reduce((prev, curr) => prev + curr, 0);

    // return results;
    return totalBytes;
});

Observe a linha yield context.df.Task.all(tasks);. O código não gera as chamadas individuais para E2_CopyFileToBlob, portanto, elas são executadas em paralelo. Quando o orquestrador passa a matriz de tarefas para context.df.Task.all, ele retorna uma tarefa que só é concluída quando todas as operações de cópia são concluídas. Se estiver familiarizado com Promise.all em JavaScript, isso não será novidade para você. A diferença é que essas tarefas podem estar em execução em várias máquinas virtuais simultaneamente e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Observação

Embora as tarefas sejam conceitualmente semelhantes a promessas JavaScript, funções de orquestrador devem usar context.df.Task.all e context.df.Task.any em vez de Promise.all e Promise.race para gerenciar a paralelização de tarefa.

Depois que o orquestrador produzir context.df.Task.all, todas as chamadas de função serão concluídas e retornarão valores. Cada chamada para E2_CopyFileToBlob retorna o número de bytes carregados, de forma que calcular a contagem total de bytes é uma questão de adicionar todos esses valores retornados.

Modelo de programação V4

Este é o código que implementa a função de orquestrador:

const df = require("durable-functions");
const path = require("path");
const getFileListActivityName = "getFileList";
const copyFileToBlobActivityName = "copyFileToBlob";

df.app.orchestration("backupSiteContent", function* (context) {
    const rootDir = context.df.getInput();
    if (!rootDir) {
        throw new Error("A directory path is required as an input.");
    }

    const rootDirAbs = path.resolve(rootDir);
    const files = yield context.df.callActivity(getFileListActivityName, rootDirAbs);

    // Backup Files and save Tasks into array
    const tasks = [];
    for (const file of files) {
        const input = {
            backupPath: path.relative(rootDirAbs, file).replace(/\\/g, "/"),
            filePath: file,
        };
        tasks.push(context.df.callActivity(copyFileToBlobActivityName, input));
    }

    // wait for all the Backup Files Activities to complete, sum total bytes
    const results = yield context.df.Task.all(tasks);
    const totalBytes = results ? results.reduce((prev, curr) => prev + curr, 0) : 0;

    // return results;
    return totalBytes;
});

Observe a linha yield context.df.Task.all(tasks);. Todas as chamadas individuais para a copyFileToBlob função não foram suspensas, o que permite que elas sejam executadas em paralelo. Quando passamos essa matriz de tarefas para context.df.Task.all, obtemos uma tarefa que não será concluída até que todas as operações de cópia tenham sido concluídas. Se estiver familiarizado com Promise.all em JavaScript, isso não será novidade para você. A diferença é que essas tarefas podem estar em execução em várias máquinas virtuais simultaneamente e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Observação

Embora as tarefas sejam conceitualmente semelhantes a promessas JavaScript, funções de orquestrador devem usar context.df.Task.all e context.df.Task.any em vez de Promise.all e Promise.race para gerenciar a paralelização de tarefa.

Depois de suspender context.df.Task.all, sabemos que todas as chamadas de função foram concluídas e retornaram valores para nós. Cada chamada para copyFileToBlob retorna o número de bytes carregados, de forma que calcular a contagem total de bytes é uma questão de adicionar todos esses valores retornados.

Python

A função usa o function.json padrão para funções de orquestrador.

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

Este é o código que implementa a função de orquestrador:

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)

Observe a linha yield context.task_all(tasks);. O código não gera as chamadas individuais para E2_CopyFileToBlob, portanto, elas são executadas em paralelo. Quando o orquestrador passa a matriz de tarefas para context.task_all, ele retorna uma tarefa que só é concluída quando todas as operações de cópia são concluídas. Se você estiver familiarizado com asyncio.gather em Python, isso não é novidade para você. A diferença é que essas tarefas podem estar em execução em várias máquinas virtuais simultaneamente e a extensão Durable Functions garante que a execução de ponta a ponta seja resiliente à reciclagem de processos.

Observação

Embora as tarefas sejam conceitualmente semelhantes a Python awaitables, as funções de orquestrador devem usar yield, bem como as APIs context.task_all e context.task_any para gerenciar a paralelização de tarefas.

Depois que o orquestrador produzir context.task_all, todas as chamadas de função serão concluídas e retornarão valores. Cada chamada para E2_CopyFileToBlob retorna o número de bytes carregados, para que possamos calcular a contagem total de bytes somando todos os valores retornados juntos.

PowerShell

Um exemplo do PowerShell ainda não está disponível.

Java

Um exemplo de Java ainda não está disponível.

O orquestrador faz o seguinte:

1. Usa uma lista de itens de trabalho como entrada.
2. Realiza fan-out, criando uma tarefa para cada item de trabalho e processando-os em paralelo.
3. Aguarda a conclusão de todas as tarefas paralelas.
4. Realiza fan-in, agregando os resultados.

C#

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

Use Task.WhenAll() para aguardar a conclusão de todas as tarefas paralelas. O SDK da Tarefa Durável garante que as tarefas possam ser executadas em várias máquinas simultaneamente e que a execução seja resiliente a reinicializações de processos.

JavaScript

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

Use whenAll() para aguardar a conclusão de todas as tarefas paralelas. O SDK da Tarefa Durável garante que as tarefas possam ser executadas em várias máquinas simultaneamente e que a execução seja resiliente a reinicializações de processos.

Python

from durabletask import task

def fan_out_fan_in_orchestrator(ctx: task.OrchestrationContext, work_items: list) -> dict:
    """Orchestrator that demonstrates 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

Use task.when_all() para aguardar a conclusão de todas as tarefas paralelas. O SDK da Tarefa Durável garante que as tarefas possam ser executadas em várias máquinas simultaneamente e que a execução seja resiliente a reinicializações de processos.

PowerShell

Este exemplo está disponível para .NET, JavaScript, Java e Python.

Java

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

Use ctx.allOf(tasks).await() para aguardar a conclusão de todas as tarefas paralelas. O SDK da Tarefa Durável garante que as tarefas possam ser executadas em várias máquinas simultaneamente e que a execução seja resiliente a reinicializações de processos.

Activities

As funções de atividade auxiliar são funções regulares que usam a activityTrigger vinculação.

Função de atividade de E2_GetFileList

C#

Modelo isolado

using System.IO;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace SampleApp;

public static class BackupSiteContent
{
    [Function("E2_GetFileList")]
    public static string[] GetFileList(
        [ActivityTrigger] string rootDirectory,
        FunctionContext executionContext)
    {
        ILogger logger = executionContext.GetLogger("E2_GetFileList");
        logger.LogInformation("Searching for files under '{RootDirectory}'...", rootDirectory);

        string[] files = Directory.GetFiles(rootDirectory, "*", SearchOption.AllDirectories);
        logger.LogInformation("Found {FileCount} file(s) under {RootDirectory}.", files.Length, rootDirectory);

        return files;
    }
}

Modelo em processo

[FunctionName("E2_GetFileList")]
public static string[] GetFileList(
    [ActivityTrigger] string rootDirectory, 
    ILogger log)
{
    log.LogInformation($"Searching for files under '{rootDirectory}'...");
    string[] files = Directory.GetFiles(rootDirectory, "*", SearchOption.AllDirectories);
    log.LogInformation($"Found {files.Length} file(s) under {rootDirectory}.");

    return files;
}

JavaScript

Modelo de programação V3

O arquivo function.json para E2_GetFileList está conforme o exemplo a seguir:

{
  "bindings": [
    {
      "name": "rootDirectory",
      "type": "activityTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

Esta é a implementação:

const readdirp = require("readdirp");

module.exports = function (context, rootDirectory) {
    context.log(`Searching for files under '${rootDirectory}'...`);
    const allFilePaths = [];

    readdirp(
        { root: rootDirectory, entryType: "all" },
        function (fileInfo) {
            if (!fileInfo.stat.isDirectory()) {
                allFilePaths.push(fileInfo.fullPath);
            }
        },
        function (err, res) {
            if (err) {
                throw err;
            }

            context.log(`Found ${allFilePaths.length} under ${rootDirectory}.`);
            context.done(null, allFilePaths);
        }
    );
};

A função usa o readdirp módulo, a versão 2.x, para ler recursivamente a estrutura do diretório.

Modelo de programação V4

Aqui está a implementação da função de atividade getFileList:

const df = require("durable-functions");
const readdirp = require("readdirp");
const getFileListActivityName = "getFileList";

df.app.activity(getFileListActivityName, {
    handler: async function (rootDirectory, context) {
        context.log(`Searching for files under '${rootDirectory}'...`);

        const allFilePaths = [];
        for await (const entry of readdirp(rootDirectory, { type: "files" })) {
            allFilePaths.push(entry.fullPath);
        }
        context.log(`Found ${allFilePaths.length} under ${rootDirectory}.`);
        return allFilePaths;
    },
});

A função utiliza o módulo readdirp (versão 3.x) para ler recursivamente a estrutura do diretório.

Python

O arquivo function.json para E2_GetFileList está conforme o exemplo a seguir:

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

Esta é a implementação:

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

PowerShell

Exemplo do PowerShell em breve.

Java

Exemplo de Java em breve.

Observação

Não coloque esse código na função de orquestrador. As funções de orquestrador não devem fazer E/S, incluindo o acesso ao sistema de arquivos local. Para obter mais informações, confira Restrições de código na função de orquestrador.

Função de atividade de E2_CopyFileToBlob

C#

Modelo isolado

Observação

Para executar o código de exemplo, instale o pacote NuGet Azure.Storage.Blobs.

using System;
using System.IO;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace SampleApp;

public static class BackupSiteContent
{
    [Function("E2_CopyFileToBlob")]
    public static async Task<long> CopyFileToBlob(
        [ActivityTrigger] string filePath,
        FunctionContext executionContext)
    {
        ILogger logger = executionContext.GetLogger("E2_CopyFileToBlob");
        long byteCount = new FileInfo(filePath).Length;

        string blobPath = filePath
            .Substring(Path.GetPathRoot(filePath)!.Length)
            .Replace('\\', '/');
        string outputLocation = $"backups/{blobPath}";

        string? connectionString = Environment.GetEnvironmentVariable("AzureWebJobsStorage");
        if (string.IsNullOrEmpty(connectionString))
        {
            throw new InvalidOperationException("AzureWebJobsStorage is not configured.");
        }

        BlobContainerClient containerClient = new(connectionString, "backups");
        await containerClient.CreateIfNotExistsAsync();
        BlobClient blobClient = containerClient.GetBlobClient(blobPath);

        logger.LogInformation("Copying '{FilePath}' to '{OutputLocation}'. Total bytes = {ByteCount}.", filePath, outputLocation, byteCount);

        await using Stream source = File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read);
        await blobClient.UploadAsync(source, overwrite: true);

        return byteCount;
    }
}

Modelo em processo

[FunctionName("E2_CopyFileToBlob")]
public static async Task<long> CopyFileToBlob(
    [ActivityTrigger] string filePath,
    Binder binder,
    ILogger log)
{
    long byteCount = new FileInfo(filePath).Length;

    // strip the drive letter prefix and convert to forward slashes
    string blobPath = filePath
        .Substring(Path.GetPathRoot(filePath).Length)
        .Replace('\\', '/');
    string outputLocation = $"backups/{blobPath}";

    log.LogInformation($"Copying '{filePath}' to '{outputLocation}'. Total bytes = {byteCount}.");

    // copy the file contents into a blob
    using (Stream source = File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read))
    using (Stream destination = await binder.BindAsync<CloudBlobStream>(
        new BlobAttribute(outputLocation, FileAccess.Write)))
    {
        await source.CopyToAsync(destination);
    }

    return byteCount;
}

Observação

O exemplo de modelo em processo requer o pacote NuGet Microsoft.Azure.WebJobs.Extensions.Storage e usa recursos de associação do Azure Functions como o Binder parâmetro.

JavaScript

Modelo de programação V3

O arquivo function.json para E2_CopyFileToBlob também é simples:

{
  "bindings": [
    {
      "name": "filePath",
      "type": "activityTrigger",
      "direction": "in"
    },
    {
      "name": "out",
      "type": "blob",
      "path": "",
      "connection": "AzureWebJobsStorage",
      "direction": "out"
    }
  ],
  "disabled": false
}

A implementação do JavaScript usa o SDK Azure Storage para Node para carregar os arquivos para Azure Blob Storage.

const fs = require("fs");
const path = require("path");
const storage = require("azure-storage");

module.exports = function (context, filePath) {
    const container = "backups";
    const root = path.parse(filePath).root;
    const blobPath = filePath.substring(root.length).replace("\\", "/");
    const outputLocation = `backups/${blobPath}`;
    const blobService = storage.createBlobService();

    blobService.createContainerIfNotExists(container, (error) => {
        if (error) {
            throw error;
        }

        fs.stat(filePath, function (error, stats) {
            if (error) {
                throw error;
            }
            context.log(
                `Copying '${filePath}' to '${outputLocation}'. Total bytes = ${stats.size}.`
            );

            const readStream = fs.createReadStream(filePath);

            blobService.createBlockBlobFromStream(
                container,
                blobPath,
                readStream,
                stats.size,
                function (error) {
                    if (error) {
                        throw error;
                    }

                    context.done(null, stats.size);
                }
            );
        });
    });
};

Modelo de programação V4

A implementação de JavaScript de copyFileToBlob usa uma associação de saída do Armazenamento do Azure para carregar os arquivos no Armazenamento de Blobs do Azure.

const df = require("durable-functions");
const fs = require("fs/promises");
const { output } = require("@azure/functions");

const copyFileToBlobActivityName = "copyFileToBlob";

const blobOutput = output.storageBlob({
    path: "backups/{backupPath}",
    connection: "StorageConnString",
});

df.app.activity(copyFileToBlobActivityName, {
    extraOutputs: [blobOutput],
    handler: async function ({ backupPath, filePath }, context) {
        const outputLocation = `backups/${backupPath}`;
        const stats = await fs.stat(filePath);
        context.log(`Copying '${filePath}' to '${outputLocation}'. Total bytes = ${stats.size}.`);

        const fileContents = await fs.readFile(filePath);

        context.extraOutputs.set(blobOutput, fileContents);

        return stats.size;
    },
});

Python

O arquivo function.json para E2_CopyFileToBlob também é simples:

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

A implementação do Python usa o SDK Azure Storage para Python para carregar os arquivos para 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

PowerShell

Exemplo do PowerShell em breve.

Java

Exemplo de Java em breve.

A implementação carrega o arquivo do disco e transmite de forma assíncrona o conteúdo para um blob com o mesmo nome no backups contêiner. A função retorna o número de bytes copiados para o armazenamento. O orquestrador usa esse valor para calcular a soma agregada.

Observação

Este exemplo move as operações de E/S para uma activityTrigger função. O trabalho pode ser executado em vários computadores e dá suporte à checagem de progresso. Se o processo de host terminar, você saberá quais uploads estão concluídos.

As atividades fazem o trabalho. Ao contrário dos orquestradores, as atividades podem executar operações de E/S e lógica não determinística.

Processar atividade do item de trabalho

C#

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 (this is where you do the actual work)
        var result = new Dictionary<string, int>
        {
            { workItem, workItem.Length }
        };

        return Task.FromResult(result);
    }
}

JavaScript

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

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

Ao contrário dos orquestradores, as atividades podem executar operações de E/S, como chamadas HTTP, consultas de banco de dados e acesso a arquivos.

Python

from durabletask import task

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

PowerShell

Este exemplo é mostrado para .NET, JavaScript, Java e Python.

Java

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

Atividade de resultados agregados

C#

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

JavaScript

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

Ao contrário dos orquestradores, as atividades podem executar operações de E/S, como chamadas HTTP, consultas de banco de dados e acesso a arquivos.

Python

from durabletask import task

def aggregate_results(ctx: task.ActivityContext, results: list) -> dict:
    """Activity that aggregates 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
    }

PowerShell

Este exemplo é mostrado para .NET, JavaScript, Java e Python.

Java

No exemplo de Java, o orquestrador agrega resultados depois do retorno de ctx.allOf(tasks).await().

Executar o exemplo

Inicie a orquestração no Windows enviando a seguinte solicitação HTTP POST:

POST http://{host}/orchestrators/E2_BackupSiteContent
Content-Type: application/json
Content-Length: 20

"D:\\home\\LogFiles"

Como alternativa, em um aplicativo de funções do Linux, inicie a orquestração enviando a seguinte solicitação HTTP POST. Python atualmente é executado no Linux para Serviço de Aplicativo:

POST http://{host}/orchestrators/E2_BackupSiteContent
Content-Type: application/json
Content-Length: 20

"/home/site/wwwroot"

Observação

A HttpStart função espera JSON. Inclua o Content-Type: application/json cabeçalho e codifique o caminho do diretório como uma cadeia de caracteres JSON. O extrato HTTP pressupõe que host.json tenha uma entrada que remove o prefixo padrão api/ de todas as URLs de função de gatilho HTTP. Encontre a marcação para esta configuração no arquivo host.json de exemplo.

Esta solicitação HTTP dispara o orquestrador E2_BackupSiteContent e passa a cadeia de caracteres D:\home\LogFiles como um parâmetro. A resposta tem um link para verificar o status da operação de backup:

HTTP/1.1 202 Accepted
Content-Length: 719
Content-Type: application/json; charset=utf-8
Location: http://{host}/runtime/webhooks/durabletask/instances/b4e9bdcc435d460f8dc008115ff0a8a9?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

(...trimmed...)

Dependendo do número de arquivos de log em seu aplicativo de funções, essa operação pode levar vários minutos para ser concluída. Obtenha o status mais recente consultando a URL no Location cabeçalho da resposta HTTP 202 anterior:

GET http://{host}/runtime/webhooks/durabletask/instances/b4e9bdcc435d460f8dc008115ff0a8a9?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

HTTP/1.1 202 Accepted
Content-Length: 148
Content-Type: application/json; charset=utf-8
Location: http://{host}/runtime/webhooks/durabletask/instances/b4e9bdcc435d460f8dc008115ff0a8a9?taskHub=DurableFunctionsHub&connection=Storage&code={systemKey}

{"runtimeStatus":"Running","input":"D:\\home\\LogFiles","output":null,"createdTime":"2019-06-29T18:50:55Z","lastUpdatedTime":"2019-06-29T18:51:16Z"}

Nesse caso, a função ainda está sendo executada. A resposta mostra a entrada que foi salva no estado do orquestrador e a hora da última atualização. Use o valor do cabeçalho Location para sondar a conclusão. Quando o status é "Concluído", a resposta se assemelha ao seguinte exemplo:

HTTP/1.1 200 OK
Content-Length: 152
Content-Type: application/json; charset=utf-8

{"runtimeStatus":"Completed","input":"D:\\home\\LogFiles","output":452071,"createdTime":"2019-06-29T18:50:55Z","lastUpdatedTime":"2019-06-29T18:51:26Z"}

A resposta mostra que a orquestração foi concluída e o tempo aproximado para concluir. O campo output indica que foram carregados cerca de 450 KB de registros pela orquestração.

Para executar o exemplo:

1. Inicie o emulador do Agendador de Tarefas Duráveis para desenvolvimento local.

   docker run -d -p 8080:8080 -p 8082:8082 --name dts-emulator mcr.microsoft.com/dts/dts-emulator:latest
   

2. Inicia o worker para registrar o orquestrador e as atividades.

3. Execute o cliente para agendar uma orquestração com uma lista de itens de trabalho:

C#

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

JavaScript

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

Crie o DurableTaskAzureManagedClientBuilder usando uma cadeia de conexão para o Agendador de Tarefas Duráveis. Use scheduleNewOrchestration para iniciar uma orquestração e use waitForOrchestrationCompletion para aguardar a conclusão.

Python

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

PowerShell

Este exemplo é mostrado para .NET, JavaScript, Java e Python.

Java

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

Próximas Etapas 

Este exemplo mostra o padrão fan-out/fan-in. O exemplo a seguir mostra como implementar o padrão de monitor com temporizadores duráveis.

Executar o exemplo do monitor

Este artigo demonstra o padrão fan-out/fan-in. Explore mais padrões e recursos:

Introdução aos SDKs de Tarefas Duráveis

Para obter exemplos do SDK do JavaScript, consulte os exemplos do SDK do JavaScript da Tarefa Durável.