Gegevenspersistentie en serialisatie in Durable Functions voor Azure Functions

Dit voorbeeld vereist de Azure. Storage.Blobs NuGet-pakket.

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

Dit voorbeeld vereist de Azure. Storage.Blobs NuGet-pakket.

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

Dit voorbeeld vereist het npm-pakket @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,
        };
    },
});

Dit voorbeeld vereist het azure-storage-blob-pakket . Voeg het toe aan requirements.txt.

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

Dit voorbeeld vereist de Az.Storage-module . Voeg het toe aan requirements.psd1 en zorg dat beheerde afhankelijkheden in host.jsonzijn ingeschakeld.

Orkestrator run.ps1:

param($Context)

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

return $OutputReference

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

Gebruik de standaard orchestrationTrigger en activityTrigger bindings beschreven in Durable Functions bindings.

Dit voorbeeld vereist de Maven-pakketten com.azure:azure-storage-blob en 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 Isolated en System.Text.Json

Durable Functions die worden uitgevoerd in het .NET Isolated-werkproces, gebruiken dezelfde objectserialisator die globaal is geconfigureerd voor je Azure Functions-app (zie WorkerOptions). Deze serializer is System.Text.Json standaard in plaats van Newtonsoft.Json. Alle wijzigingen in WorkerOptions.Serializer zijn transitief ook van toepassing op Durable Functions.

Zie de JSON-serialisatie en deserialisatie in .NET overzichtsdocumentatie voor meer informatie over de ingebouwde ondersteuning voor JSON-serialisatie in .NET.

Standaardserialisatielogica

Durable Functions voor .NET maakt intern gebruik van Json.NET om orkestratie- en entiteitsgegevens naar JSON te serialiseren. De standaardinstellingen die worden gebruikt voor Json.NET zijn:

Invoer, uitvoer en status:

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

Uitzonderingen:

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

Lees hier meer gedetailleerde documentatie.JsonSerializerSettings

Serialisatie aanpassen met .NET kenmerken

Tijdens serialisatie zoekt Json.NET naar verschillende kenmerken op klassen en eigenschappen die bepalen hoe de gegevens worden geserialiseerd en gedeserialiseerd vanuit JSON. Als u eigenaar bent van de broncode voor het gegevenstype dat wordt doorgegeven aan Durable Functions API's, kunt u deze kenmerken toevoegen aan het type om serialisatie en deserialisatie aan te passen.

Pas de serialisatie aan met injectie van afhankelijkheden

Functie-apps die zijn gericht op .NET en worden uitgevoerd op de Functions V3-runtime, kunnen gebruikmaken van Dependency Injection (DI) om aan te passen hoe gegevens en uitzonderingen worden geserialiseerd. In de volgende voorbeeldcode ziet u hoe u DI gebruikt om de standaard-Json.NET serialisatie-instellingen te overschrijven met behulp van aangepaste implementaties van de IMessageSerializerSettingsFactory en IErrorSerializerSettingsFactory-serviceinterfaces.

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

Serialisatie- en deserialisatielogica

Azure Functions Node-toepassingen gebruiken JSON.stringify() voor serialisatie en JSON.Parse() voor deserialisatie. De meeste types moeten naadloos serialiseren en deserialiseren. In gevallen waarin de standaardlogica onvoldoende is, overschrijft het definiëren van een toJSON() methode voor het object de serialisatielogica. Er bestaat echter geen analogie voor objectdeserialisatie.

Voor volledige aanpassing van de serialisatie-/deserialisatiepijplijn kunt u de serialisatie en deserialisatie afhandelen met uw eigen code en gegevens doorgeven als tekenreeksen.

Serialisatie- en deserialisatielogica

Durable Functions PowerShell-apps gebruiken de standaard PowerShell JSON-afhandeling voor orkestratie- en activiteitspayloads. In de meeste gevallen serialiseren en deserialiseren gewone JSON-compatibele objecten correct zonder extra configuratie.

Als je aangepast gedrag nodig hebt voor een payloadvorm, serialiseer of deserialiseer dan de data zelf binnen je functies en geef de resulterende JSON-compatibele waarden door tussen orchestrators, activiteiten en entiteiten. Deze aanpak houdt het dataformaat expliciet en voorkomt het gebruik van een serializer-aanpassings-API die niet beschikbaar is voor PowerShell.

Voor complexere scenario's kun je overwegen je data te representeren als eenvoudige strings, arrays of hashtables en het converteren tussen formaten binnen je functiecode.

Serialisatie- en deserialisatielogica

Het is raadzaam om typeaantekeningen te gebruiken om ervoor te zorgen dat Durable Functions uw gegevens correct serialiseert en deserialiseerd. Hoewel veel ingebouwde typen automatisch worden verwerkt, vereisen sommige ingebouwde gegevenstypen typeaantekeningen om het type tijdens deserialisatie te behouden.

Voor aangepaste gegevenstypen maakt u JSON-serialisatie en deserialisatie mogelijk door klassemethoden to_json en from_json op uw gegevenstypeklasse te definiëren. Merk op dat deze methoden niet worden aangeroepen op de retourwaarde van de orchestratorfunctie, wat betekent dat de retourwaarde native JSON-serialiseerbaar moet zijn. Voor meer informatie, zie Bindingen.

Zie Migrate to type-safe serialization in Durable Functions for Python om gedeserialiseerde payloads te valideren aan de hand van een verwacht type en desgewenst te kiezen voor een strengere, beveiligde modus.

Java maakt gebruik van de bibliotheken Jackson v2.x voor serialisatie en deserialisatie van gegevenspayloads. U kunt Jackson-aantekeningen op uw POJO-typen gebruiken om het serialisatiegedrag aan te passen.