Datenpersistenz und Serialisierung in Durable Functions für Azure Functions

Dieses Beispiel benötigt das Azure. Storage.Blobs NuGet-Paket.

using System;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

public record BlobReference(string Container, string BlobName);

public static class LargePayloadFunctions
{
    [Function("ProcessLargePayload")]
    public static async Task<BlobReference> RunOrchestrator(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        BlobReference inputReference = context.GetInput<BlobReference>()
            ?? throw new InvalidOperationException("A blob reference is required.");

        return await context.CallActivityAsync<BlobReference>(
            nameof(ProcessLargePayloadActivity), inputReference);
    }

    [Function(nameof(ProcessLargePayloadActivity))]
    public static async Task<BlobReference> ProcessLargePayloadActivity(
        [ActivityTrigger] BlobReference inputReference)
    {
        string connectionString =
            Environment.GetEnvironmentVariable("PAYLOAD_STORAGE_CONNECTION_STRING")
            ?? throw new InvalidOperationException("Payload storage is not configured.");
        BlobServiceClient service = new BlobServiceClient(connectionString);

        BlobClient inputBlob = service
            .GetBlobContainerClient(inputReference.Container)
            .GetBlobClient(inputReference.BlobName);
        BinaryData inputData = (await inputBlob.DownloadContentAsync()).Value.Content;

        BlobContainerClient outputContainer =
            service.GetBlobContainerClient("processed-payloads");
        await outputContainer.CreateIfNotExistsAsync();

        string outputName = $"processed/{Guid.NewGuid():N}.json";
        await outputContainer.GetBlobClient(outputName)
            .UploadAsync(inputData, overwrite: true);

        return new BlobReference(outputContainer.Name, outputName);
    }
}

Dieses Beispiel benötigt das Azure. Storage.Blobs NuGet-Paket.

using System;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;

public class BlobReference
{
    public string Container { get; set; } = string.Empty;
    public string BlobName { get; set; } = string.Empty;
}

public static class LargePayloadFunctions
{
    [FunctionName("ProcessLargePayload")]
    public static async Task<BlobReference> RunOrchestrator(
        [OrchestrationTrigger] IDurableOrchestrationContext context)
    {
        BlobReference inputReference = context.GetInput<BlobReference>()
            ?? throw new InvalidOperationException("A blob reference is required.");

        return await context.CallActivityAsync<BlobReference>(
            "ProcessLargePayloadActivity", inputReference);
    }

    [FunctionName("ProcessLargePayloadActivity")]
    public static async Task<BlobReference> RunActivity(
        [ActivityTrigger] BlobReference inputReference)
    {
        string connectionString =
            Environment.GetEnvironmentVariable("PAYLOAD_STORAGE_CONNECTION_STRING")
            ?? throw new InvalidOperationException("Payload storage is not configured.");
        BlobServiceClient service = new BlobServiceClient(connectionString);

        BlobClient inputBlob = service
            .GetBlobContainerClient(inputReference.Container)
            .GetBlobClient(inputReference.BlobName);
        BinaryData inputData = (await inputBlob.DownloadContentAsync()).Value.Content;

        BlobContainerClient outputContainer =
            service.GetBlobContainerClient("processed-payloads");
        await outputContainer.CreateIfNotExistsAsync();

        string outputName = $"processed/{Guid.NewGuid():N}.json";
        await outputContainer.GetBlobClient(outputName)
            .UploadAsync(inputData, overwrite: true);

        return new BlobReference
        {
            Container = outputContainer.Name,
            BlobName = outputName,
        };
    }
}

Dieses Beispiel benötigt das npm-Paket @azure/storage-blob .

const { randomUUID } = require("crypto");
const { BlobServiceClient } = require("@azure/storage-blob");
const df = require("durable-functions");

df.app.orchestration("processLargePayload", function* (context) {
    const inputReference = context.df.getInput();
    return yield context.df.callActivity(
        "processLargePayloadActivity", inputReference);
});

df.app.activity("processLargePayloadActivity", {
    handler: async (inputReference) => {
        const connectionString =
            process.env.PAYLOAD_STORAGE_CONNECTION_STRING;
        if (!connectionString) {
            throw new Error(
                "PAYLOAD_STORAGE_CONNECTION_STRING is not set.");
        }

        const service = BlobServiceClient.fromConnectionString(
            connectionString);

        const inputBlob = service
            .getContainerClient(inputReference.container)
            .getBlockBlobClient(inputReference.blobName);
        const inputData = await inputBlob.downloadToBuffer();

        const outputContainer =
            service.getContainerClient("processed-payloads");
        await outputContainer.createIfNotExists();

        const outputName = `processed/${randomUUID()}.json`;
        await outputContainer.getBlockBlobClient(outputName)
            .uploadData(inputData);

        return {
            container: outputContainer.containerName,
            blobName: outputName,
        };
    },
});

Dieses Beispiel benötigt das azure-storage-blob-Paket . Füge es zu requirements.txt hinzu.

import os
import uuid

import azure.functions as func
import azure.durable_functions as df
from azure.core.exceptions import ResourceExistsError
from azure.storage.blob import BlobServiceClient

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.orchestration_trigger(context_name="context")
def process_large_payload(context: df.DurableOrchestrationContext):
    input_reference: dict = context.get_input()
    return (yield context.call_activity(
        "process_large_payload_activity", input_reference))


@app.activity_trigger(input_name="input_reference")
def process_large_payload_activity(input_reference: dict) -> dict:
    connection_string = os.getenv("PAYLOAD_STORAGE_CONNECTION_STRING")
    if not connection_string:
        raise ValueError(
            "PAYLOAD_STORAGE_CONNECTION_STRING is not set.")

    service = BlobServiceClient.from_connection_string(connection_string)

    input_blob = service.get_blob_client(
        container=input_reference["container"],
        blob=input_reference["blobName"])
    input_data = input_blob.download_blob().readall()

    output_container = service.get_container_client("processed-payloads")
    try:
        output_container.create_container()
    except ResourceExistsError:
        pass

    output_name = f"processed/{uuid.uuid4()}.json"
    output_container.upload_blob(
        name=output_name, data=input_data, overwrite=True)

    return {
        "container": output_container.container_name,
        "blobName": output_name,
    }

Dieses Beispiel benötigt das Az.Storage-Modul . Füge es zu requirements.psd1 hinzu und stelle sicher, dass verwaltete Abhängigkeiten in host.jsonaktiviert sind.

Orchestrator run.ps1:

param($Context)

$OutputReference = Invoke-DurableActivity `
    -FunctionName 'ProcessLargePayloadActivity' `
    -Input $Context.Input

return $OutputReference

Aktivität run.ps1:

param($InputReference)

if ([string]::IsNullOrEmpty(
        $env:PAYLOAD_STORAGE_CONNECTION_STRING)) {
    throw 'The PAYLOAD_STORAGE_CONNECTION_STRING app setting is not configured.'
}

$StorageContext = New-AzStorageContext `
    -ConnectionString $env:PAYLOAD_STORAGE_CONNECTION_STRING
$InputFile = New-TemporaryFile

try {
    Get-AzStorageBlobContent `
        -Container $InputReference.container `
        -Blob $InputReference.blobName `
        -Destination $InputFile.FullName `
        -Context $StorageContext `
        -Force | Out-Null

    # Process $InputFile here. This example uploads it unchanged.
    $OutputContainer = 'processed-payloads'
    New-AzStorageContainer `
        -Name $OutputContainer `
        -Context $StorageContext `
        -ErrorAction SilentlyContinue | Out-Null

    $OutputName = "processed/$([guid]::NewGuid()).json"
    Set-AzStorageBlobContent `
        -File $InputFile.FullName `
        -Container $OutputContainer `
        -Blob $OutputName `
        -Context $StorageContext `
        -Force | Out-Null

    return @{
        container = $OutputContainer
        blobName = $OutputName
    }
}
finally {
    Remove-Item $InputFile.FullName -Force
}

Verwenden Sie die standardmäßigen orchestrationTrigger- und activityTrigger-Bindungen, die in Durable Functions-Bindungen beschrieben werden.

Dieses Beispiel erfordert die Maven-Pakete com.azure:azure-storage-blob und com.microsoft:durabletask-azure-functions.

import com.azure.core.util.BinaryData;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobServiceClient;
import com.azure.storage.blob.BlobServiceClientBuilder;
import com.microsoft.azure.functions.annotation.FunctionName;
import com.microsoft.durabletask.TaskOrchestrationContext;
import com.microsoft.durabletask.azurefunctions.DurableActivityTrigger;
import com.microsoft.durabletask.azurefunctions.DurableOrchestrationTrigger;

import java.util.Objects;
import java.util.UUID;

public class LargePayloadFunctions {
    public static class BlobReference {
        public String container;
        public String blobName;

        public BlobReference() {
        }

        public BlobReference(String container, String blobName) {
            this.container = container;
            this.blobName = blobName;
        }
    }

    @FunctionName("ProcessLargePayload")
    public BlobReference runOrchestrator(
            @DurableOrchestrationTrigger(name = "ctx")
            TaskOrchestrationContext ctx) {
        BlobReference inputReference = Objects.requireNonNull(
            ctx.getInput(BlobReference.class),
            "A blob reference is required.");
        return ctx.callActivity(
            "ProcessLargePayloadActivity",
            inputReference,
            BlobReference.class).await();
    }

    @FunctionName("ProcessLargePayloadActivity")
    public BlobReference runActivity(
            @DurableActivityTrigger(name = "inputReference")
            BlobReference inputReference) {
        String connectionString = Objects.requireNonNull(
            System.getenv("PAYLOAD_STORAGE_CONNECTION_STRING"),
            "PAYLOAD_STORAGE_CONNECTION_STRING is not configured.");
        BlobServiceClient service = new BlobServiceClientBuilder()
            .connectionString(connectionString)
            .buildClient();

        BinaryData inputData = service
            .getBlobContainerClient(inputReference.container)
            .getBlobClient(inputReference.blobName)
            .downloadContent();

        BlobContainerClient outputContainer =
            service.getBlobContainerClient("processed-payloads");
        outputContainer.createIfNotExists();

        String outputName =
            "processed/" + UUID.randomUUID() + ".json";
        outputContainer.getBlobClient(outputName)
            .upload(inputData, true);

        return new BlobReference(
            outputContainer.getBlobContainerName(), outputName);
    }
}

.NET (isoliert) und System.Text.Json

Durable Functions, die im .NET Isolated Worker-Prozess laufen, verwenden denselben Objektserialisierer, der global für Ihre Azure Functions-App konfiguriert ist (siehe WorkerOptions). Dieser Serialisierer ist System.Text.Json standardmäßig anstelle von Newtonsoft.Json. Alle Änderungen an WorkerOptions.Serializer wirken sich transitiv auf Durable Functions aus.

Weitere Informationen zur integrierten Unterstützung der JSON-Serialisierung in .NET finden Sie in der JSON-Serialisierung und Deserialisierung in .NET Übersichtsdokumentation.

Standardserialisierungslogik

Durable Functions für .NET verwendet im Prozess Json.NET zum Serialisieren von Orchestrierungs- und Entitätsdaten in JSON. Standardmäßig werden die folgenden Json.NET-Einstellungen verwendet:

Eingaben, Ausgaben und Zustand:

JsonSerializerSettings
{
    TypeNameHandling = TypeNameHandling.None,
    DateParseHandling = DateParseHandling.None,
}

Ausnahmen:

JsonSerializerSettings
{
    ContractResolver = new ExceptionResolver(),
    TypeNameHandling = TypeNameHandling.Objects,
    ReferenceLoopHandling = ReferenceLoopHandling.Ignore,
}

Eine ausführlichere Dokumentation zu JsonSerializerSettings finden Sie hier.

Anpassen der Serialisierung mit .NET Attributen

Während der Serialisierung sucht Json.NET nach verschiedenen Attributen an Klassen und Eigenschaften, die steuern, wie die Daten serialisiert und aus JSON deserialisiert werden. Wenn Sie den Quellcode für den an die Durable Functions-APIs übergebenen Datentyp besitzen, sollten Sie dem Typ diese Attribute hinzufügen, um Serialisierung und Deserialisierung anzupassen.

Anpassen der Serialisierung mit Abhängigkeitsinjektion

Funktions-Apps, die auf .NET abzielen und auf der Funktions-V3-Laufzeit ausgeführt werden, können Dependency Injection (DI) verwenden, um anzupassen, wie Daten und Ausnahmen serialisiert werden. Im folgenden Beispielcode wird die Verwendung von DI zum Überschreiben der Standard-Json.NET Serialisierungseinstellungen mithilfe von benutzerdefinierten Implementierungen der IMessageSerializerSettingsFactory und IErrorSerializerSettingsFactory-Dienstschnittstellen veranschaulicht.

using Microsoft.Azure.Functions.Extensions.DependencyInjection;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json;
using System.Collections.Generic;

[assembly: FunctionsStartup(typeof(MyApplication.Startup))]
namespace MyApplication
{
    public class Startup : FunctionsStartup
    {
        public override void Configure(IFunctionsHostBuilder builder)
        {
            builder.Services.AddSingleton<IMessageSerializerSettingsFactory, CustomMessageSerializerSettingsFactory>();
            builder.Services.AddSingleton<IErrorSerializerSettingsFactory, CustomErrorSerializerSettingsFactory>();
        }

        /// <summary>
        /// A factory that provides the serialization for all inputs and outputs for activities and
        /// orchestrations, as well as entity state.
        /// </summary>
        internal class CustomMessageSerializerSettingsFactory : IMessageSerializerSettingsFactory
        {
            public JsonSerializerSettings CreateJsonSerializerSettings()
            {
                // Return your custom JsonSerializerSettings here
            }
        }

        /// <summary>
        /// A factory that provides the serialization for all exceptions thrown by activities
        /// and orchestrations
        /// </summary>
        internal class CustomErrorSerializerSettingsFactory : IErrorSerializerSettingsFactory
        {
            public JsonSerializerSettings CreateJsonSerializerSettings()
            {
                // Return your custom JsonSerializerSettings here
            }
        }
    }
}

Serialisierungs- und Deserialisierungslogik

Azure Functions Node-Anwendungen verwenden JSON.stringify() für die Serialisierung und JSON.Parse() für deserialisierung. Die meisten Typen sollten nahtlos serialisiert und deserialisiert werden. In Fällen, in denen die Standardlogik nicht ausreicht, setzt das Definieren einer toJSON() Methode für das Objekt die Serialisierungslogik außer Kraft. Jedoch existiert für die Objektdeserialisierung kein entsprechendes Analogon.

Für eine vollständige Anpassung der Serialisierungs-/Deserialisierungspipeline sollten Sie die Serialisierung und Deserialisierung mit Ihrem eigenen Code verarbeiten und die Daten dabei als Zeichenfolgen übergeben.

Serialisierungs- und Deserialisierungslogik

Durable Functions PowerShell-Apps verwenden die Standard-PowerShell-JSON-Behandlung für Orchestrierungs- und Aktivitätsnutzlasten. In den meisten Fällen serialisieren und deserialisieren einfache JSON-kompatible Objekte korrekt ohne zusätzliche Konfiguration.

Wenn Sie benutzerdefiniertes Verhalten für eine Nutzlastform benötigen, serialisieren oder deserialisieren Sie die Daten selbst innerhalb Ihrer Funktionen und leiten die resultierenden JSON-kompatiblen Werte zwischen Orchestratoren, Aktivitäten und Entitäten weiter. Dieser Ansatz hält das Datenformat explizit und vermeidet die Abhängigkeit von einer Serializer-Anpassungs-API, die für PowerShell nicht verfügbar ist.

Für komplexere Szenarien sollten Sie Ihre Daten als einfache Strings, Arrays oder Hashtabellen darstellen und zwischen den Formaten innerhalb Ihres Funktionscodes konvertieren.

Serialisierungs- und Deserialisierungslogik

Es wird empfohlen, Typanmerkungen zu verwenden, um sicherzustellen, dass Durable Functions Ihre Daten ordnungsgemäß serialisiert und deserialisiert. Obwohl viele integrierte Typen automatisch verarbeitet werden, benötigen einige integrierte Datentypen Typanmerkungen, um den Typ während der Deserialisierung beizubehalten.

Für benutzerdefinierte Datentypen machen Sie die JSON-Serialisierung und -Deserialisierung möglich, indem Sie die Klassenmethoden to_json und from_json in Ihrer Datentypklasse definieren. Beachten Sie, dass diese Methoden nicht für den Rückgabewert der Orchestratorfunktion aufgerufen werden, was bedeutet, dass der Rückgabewert nativ JSON-serialisierbar sein muss. Weitere Informationen finden Sie unter "Bindungen".

Um deserialisierte Nutzdaten anhand eines erwarteten Typs zu überprüfen und optional einen gehärteten strikten Modus zu aktivieren, siehe Migrate to type-safe serialization in Durable Functions for Python.

Java verwendet die Jackson v2.x-Bibliotheken für die Serialisierung und Deserialisierung von Datennutzlasten. Sie können Jackson-Anmerkungen für Ihre POJO-Typen verwenden, um das Serialisierungsverhalten anzupassen.