Trigger služby Azure Event Hubs pro Azure Functions

  • Článek

Tento článek vysvětluje, jak pracovat s triggerem služby Azure Event Hubs pro Azure Functions. Azure Functions podporuje aktivační a výstupní vazby pro službu Event Hubs.

Informace o nastavení a konfiguraci najdete v přehledu.

Pomocí triggeru funkce můžete reagovat na událost odeslanou do streamu událostí centra událostí. K nastavení triggeru musíte mít přístup pro čtení k podkladovému centru událostí. Při aktivaci funkce se zpráva předaná funkci zadá jako řetězec.

Rozhodnutí o škálování služby Event Hubs pro plány Consumption a Premium se provádějí prostřednictvím cílového škálování. Další informace najdete v tématu Škálování na základě cíle.

Informace o tom, jak Azure Functions reaguje na události odeslané do streamu událostí centra událostí pomocí triggerů, najdete v tématu Integrace služby Event Hubs s bezserverovými funkcemi v Azure.

Důležité

Tento článek používá karty pro podporu více verzí programovacího modelu Node.js. Model v4 je obecně dostupný a je navržený tak, aby měl flexibilnější a intuitivnější prostředí pro vývojáře v JavaScriptu a TypeScriptu. Další podrobnosti o tom, jak model v4 funguje, najdete v příručce pro vývojáře služby Azure Functions Node.js. Další informace o rozdílech mezi v3 a v4 najdete v průvodci migrací.

Azure Functions podporuje dva programovací modely pro Python. Způsob, jakým definujete vazby, závisí na zvoleném programovacím modelu.

Programovací model Pythonu v2 umožňuje definovat vazby pomocí dekorátorů přímo v kódu funkce Pythonu. Další informace najdete v příručce pro vývojáře Pythonu.

Tento článek podporuje oba programovací modely.

Příklad

Následující příklad ukazuje funkci jazyka C#, která se aktivuje na základě centra událostí, kde je vstupní řetězec zprávy zapsán do protokolů:

{
    private readonly ILogger<EventHubsFunction> _logger;

    public EventHubsFunction(ILogger<EventHubsFunction> logger)
    {
        _logger = logger;
    }

    [Function(nameof(EventHubFunction))]
    [FixedDelayRetry(5, "00:00:10")]
    [EventHubOutput("dest", Connection = "EventHubConnection")]
    public string EventHubFunction(
        [EventHubTrigger("src", Connection = "EventHubConnection")] string[] input,
        FunctionContext context)
    {
        _logger.LogInformation("First Event Hubs triggered message: {msg}", input[0]);

        var message = $"Output message created at {DateTime.Now}";
        return message;
    }

Následující příklad ukazuje funkci TypeScript triggeru Event Hubs. Funkce čte metadata událostí a protokoluje zprávu.

import { app, InvocationContext } from '@azure/functions';

export async function eventHubTrigger1(message: unknown, context: InvocationContext): Promise<void> {
    context.log('Event hub function processed message:', message);
    context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
    context.log('SequenceNumber =', context.triggerMetadata.sequenceNumber);
    context.log('Offset =', context.triggerMetadata.offset);
}

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'one',
    handler: eventHubTrigger1,
});

Pokud chcete přijímat události v dávce, nastavte cardinality na manyhodnotu , jak je znázorněno v následujícím příkladu.

import { app, InvocationContext } from '@azure/functions';

export async function eventHubTrigger1(messages: unknown[], context: InvocationContext): Promise<void> {
    context.log(`Event hub function processed ${messages.length} messages`);
    for (let i = 0; i < messages.length; i++) {
        context.log('Event hub message:', messages[i]);
        context.log(`EnqueuedTimeUtc = ${context.triggerMetadata.enqueuedTimeUtcArray[i]}`);
        context.log(`SequenceNumber = ${context.triggerMetadata.sequenceNumberArray[i]}`);
        context.log(`Offset = ${context.triggerMetadata.offsetArray[i]}`);
    }
}

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'many',
    handler: eventHubTrigger1,
});

Následující příklad ukazuje funkci JavaScriptu triggeru služby Event Hubs. Funkce čte metadata událostí a protokoluje zprávu.

const { app } = require('@azure/functions');

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'one',
    handler: (message, context) => {
        context.log('Event hub function processed message:', message);
        context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
        context.log('SequenceNumber =', context.triggerMetadata.sequenceNumber);
        context.log('Offset =', context.triggerMetadata.offset);
    },
});

Pokud chcete přijímat události v dávce, nastavte cardinality na manyhodnotu , jak je znázorněno v následujícím příkladu.

const { app } = require('@azure/functions');

app.eventHub('eventHubTrigger1', {
    connection: 'myEventHubReadConnectionAppSetting',
    eventHubName: 'MyEventHub',
    cardinality: 'many',
    handler: (messages, context) => {
        context.log(`Event hub function processed ${messages.length} messages`);
        for (let i = 0; i < messages.length; i++) {
            context.log('Event hub message:', messages[i]);
            context.log(`EnqueuedTimeUtc = ${context.triggerMetadata.enqueuedTimeUtcArray[i]}`);
            context.log(`SequenceNumber = ${context.triggerMetadata.sequenceNumberArray[i]}`);
            context.log(`Offset = ${context.triggerMetadata.offsetArray[i]}`);
        }
    },
});

Tady je kód PowerShellu:

param($eventHubMessages, $TriggerMetadata)

Write-Host "PowerShell eventhub trigger function called for message array: $eventHubMessages"

$eventHubMessages | ForEach-Object { Write-Host "Processed message: $_" }

Následující příklad ukazuje vazbu triggeru služby Event Hubs a funkci Pythonu, která tuto vazbu používá. Funkce čte metadata událostí a protokoluje zprávu. Tento příklad závisí na tom, jestli používáte programovací model v1 nebo v2 Pythonu.

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="EventHubTrigger1")
@app.event_hub_message_trigger(arg_name="myhub", 
                               event_hub_name="<EVENT_HUB_NAME>",
                               connection="<CONNECTION_SETTING>") 
def test_function(myhub: func.EventHubEvent):
    logging.info('Python EventHub trigger processed an event: %s',
                myhub.get_body().decode('utf-8'))

Následující příklad ukazuje vazbu triggeru Event Hubs, která protokoluje text zprávy triggeru Event Hubs.

@FunctionName("ehprocessor")
public void eventHubProcessor(
  @EventHubTrigger(name = "msg",
                  eventHubName = "myeventhubname",
                  connection = "myconnvarname") String message,
       final ExecutionContext context )
       {
          context.getLogger().info(message);
 }

V knihovně modulu runtime funkcí Java použijte poznámku EventHubTrigger k parametrům, jejichž hodnota pochází z centra událostí. Parametry s těmito poznámkami způsobí, že se funkce spustí při příchodu události. Tuto poznámku lze použít s nativními typy Javy, POJOs nebo hodnotami null s použitím Optional<T>.

Následující příklad ilustruje rozsáhlé použití SystemProperties a další možnosti vazby pro další introspekci události spolu s poskytováním dobře formátované BlobOutput cesty, která je hierarchický date.

package com.example;
import java.util.Map;
import java.time.ZonedDateTime;

import com.microsoft.azure.functions.annotation.*;
import com.microsoft.azure.functions.*;

/**
 * Azure Functions with Event Hub trigger.
 * and Blob Output using date in path along with message partition ID
 * and message sequence number from EventHub Trigger Properties
 */
public class EventHubReceiver {

    @FunctionName("EventHubReceiver")
    @StorageAccount("bloboutput")

    public void run(
            @EventHubTrigger(name = "message",
                eventHubName = "%eventhub%",
                consumerGroup = "%consumergroup%",
                connection = "eventhubconnection",
                cardinality = Cardinality.ONE)
            String message,

            final ExecutionContext context,

            @BindingName("Properties") Map<String, Object> properties,
            @BindingName("SystemProperties") Map<String, Object> systemProperties,
            @BindingName("PartitionContext") Map<String, Object> partitionContext,
            @BindingName("EnqueuedTimeUtc") Object enqueuedTimeUtc,

            @BlobOutput(
                name = "outputItem",
                path = "iotevents/{datetime:yy}/{datetime:MM}/{datetime:dd}/{datetime:HH}/" +
                       "{datetime:mm}/{PartitionContext.PartitionId}/{SystemProperties.SequenceNumber}.json")
            OutputBinding<String> outputItem) {

        var et = ZonedDateTime.parse(enqueuedTimeUtc + "Z"); // needed as the UTC time presented does not have a TZ
                                                             // indicator
        context.getLogger().info("Event hub message received: " + message + ", properties: " + properties);
        context.getLogger().info("Properties: " + properties);
        context.getLogger().info("System Properties: " + systemProperties);
        context.getLogger().info("partitionContext: " + partitionContext);
        context.getLogger().info("EnqueuedTimeUtc: " + et);

        outputItem.setValue(message);
    }
}

Atributy

Knihovny C# v procesu i izolovaného pracovního procesu používají ke konfiguraci triggeru atribut. Skript jazyka C# místo toho používá konfigurační soubor function.json, jak je popsáno v průvodci skriptováním jazyka C#.

EventHubTriggerAttribute Použijte k definování triggeru v centru událostí, který podporuje následující vlastnosti.

Parametry Popis
EventHubName Název centra událostí. Pokud se v připojovací řetězec nachází také název centra událostí, tato hodnota přepíše tuto vlastnost za běhu. Odkazovat můžete v nastavení aplikace, například %eventHubName%
Skupina příjemců Volitelná vlastnost, která nastaví skupinu příjemců použitou k přihlášení k odběru událostí v centru. Pokud tuto možnost vynecháte, $Default použije se skupina příjemců.
Připojení Název nastavení aplikace nebo nastavení kolekce, která určuje, jak se připojit ke službě Event Hubs. Další informace najdete v tématu Připojení iony.

Dekoratéry

Platí pouze pro programovací model Pythonu v2.

Pro funkce Pythonu v2 definované pomocí dekorátoru následující vlastnosti:cosmos_db_trigger

Vlastnost Popis
arg_name Název proměnné, která představuje položku události v kódu funkce.
event_hub_name Název centra událostí. Pokud se v připojovací řetězec nachází také název centra událostí, tato hodnota přepíše tuto vlastnost za běhu.
connection Název nastavení aplikace nebo nastavení kolekce, která určuje, jak se připojit ke službě Event Hubs. Viz Připojení iony.

Informace o funkcích Pythonu definovaných pomocí function.json najdete v části Konfigurace .

Poznámky

V knihovně modulu runtime funkcí Java použijte poznámku EventHubTrigger, která podporuje následující nastavení:

Konfigurace

Platí pouze pro programovací model Pythonu v1.

Následující tabulka vysvětluje vlastnosti, které můžete nastavit u objektu předaného options metodě app.eventHub() .

Vlastnost Popis
eventHubName Název centra událostí. Pokud se v připojovací řetězec nachází také název centra událostí, tato hodnota přepíše tuto vlastnost za běhu. Odkazovat můžete prostřednictvím nastavení aplikace.%eventHubName%
consumerGroup Volitelná vlastnost, která nastaví skupinu příjemců použitou k přihlášení k odběru událostí v centru. Pokud tuto možnost vynecháte, $Default použije se skupina příjemců.
Mohutnost Pokud chcete povolit dávkování, nastavte na many hodnotu. Pokud tuto hodnotu vynecháte nebo nastavíte one, předá se funkci jedna zpráva.
Připojení Název nastavení aplikace nebo nastavení kolekce, která určuje, jak se připojit ke službě Event Hubs. Viz Připojení iony.

Následující tabulka vysvětluje vlastnosti konfigurace triggeru, které jste nastavili v souboru function.json , které se liší podle verze modulu runtime.

vlastnost function.json Popis
type Musí být nastavena na eventHubTriggerhodnotu . Tato vlastnost se nastaví automaticky při vytváření triggeru na webu Azure Portal.
direction Musí být nastavena na inhodnotu . Tato vlastnost se nastaví automaticky při vytváření triggeru na webu Azure Portal.
Jméno Název proměnné, která představuje položku události v kódu funkce.
eventHubName Název centra událostí. Pokud se v připojovací řetězec nachází také název centra událostí, tato hodnota přepíše tuto vlastnost za běhu. Odkazovat můžete prostřednictvím nastavení aplikace.%eventHubName%
consumerGroup Volitelná vlastnost, která nastaví skupinu příjemců použitou k přihlášení k odběru událostí v centru. Pokud tuto možnost vynecháte, $Default použije se skupina příjemců.
Mohutnost Pokud chcete povolit dávkování, nastavte na many hodnotu. Pokud tuto hodnotu vynecháte nebo nastavíte one, předá se funkci jedna zpráva.
Připojení Název nastavení aplikace nebo nastavení kolekce, která určuje, jak se připojit ke službě Event Hubs. Viz Připojení iony.

Při místním vývoji přidejte nastavení aplikace do souboru local.settings.json v kolekci Values .

Využití

Další informace o tom, jak se triggery Event Hubs a triggery Služby IoT Hub škálují, najdete v tématu Využívání událostí pomocí Azure Functions.

Typ parametru podporovaný výstupní vazbou služby Event Hubs závisí na verzi modulu runtime služby Functions, verzi balíčku rozšíření a použitém režimu jazyka C#.

Pokud chcete, aby funkce zpracovávala jednu událost, trigger Event Hubs může svázat s následujícími typy:

Typ Popis
string Událost jako řetězec. Použije se, když je událost jednoduchý text.
byte[] Bajty události.
Serializovatelné typy JSON Když událost obsahuje data JSON, functions se pokusí deserializovat data JSON do prostého typu objektu CLR (POCO).
Azure.Messaging.EventHubs.EventData1 Objekt události.
Pokud migrujete ze starších verzí sad SDK služby Event Hubs, mějte na paměti, že tato verze zahodí podporu staršího Bodytypu ve prospěch EventBody.

Pokud chcete, aby funkce zpracovávala dávku událostí, trigger Event Hubs může svázat s následujícími typy:

Typ Popis
string[] Pole událostí z dávky jako řetězce. Každá položka představuje jednu událost.
EventData[]1 Pole událostí z dávky, jako instance Azure.Messaging.EventHubs.EventData. Každá položka představuje jednu událost.
T[] where T je serializovatelný typJSON 1 Pole událostí z dávky jako instance vlastního typu POCO. Každá položka představuje jednu událost.

1 Pokud chcete použít tyto typy, musíte odkazovat na Microsoft.Azure.Functions.Worker.Extensions.EventHubs 5.5.0 nebo novější a běžné závislosti pro vazby typu sady SDK.

Typ parametru může být jeden z následujících:

  • Všechny nativní typy Javy, jako je int, String, byte[].
  • Hodnoty s možnou hodnotou null pomocí volitelného.
  • Libovolný typ POJO.

Další informace najdete v referenčních informacích k EventHubTriggeru.

Metadata událostí

Trigger služby Event Hubs poskytuje několik vlastností metadat. Vlastnosti metadat lze použít jako součást vazebních výrazů v jiných vazbách nebo jako parametry v kódu. Vlastnosti pocházejí z třídy EventData .

Vlastnost Type Popis
PartitionContext PartitionContext Instance PartitionContext.
EnqueuedTimeUtc DateTime Čas zařazení do fronty ve standardu UTC
Offset string Posun dat vzhledem k datovému proudu oddílu centra událostí. Posun je značka nebo identifikátor události v rámci datového proudu služby Event Hubs. Identifikátor je jedinečný v rámci oddílu datového proudu služby Event Hubs.
PartitionKey string Oddíl, do kterého se mají odesílat data události.
Properties IDictionary<String,Object> Vlastnosti uživatele dat události.
SequenceNumber Int64 Logické pořadové číslo události.
SystemProperties IDictionary<String,Object> Systémové vlastnosti, včetně dat události.

Podívejte se na příklady kódu, které tyto vlastnosti používají dříve v tomto článku.

Propojení

Vlastnost connection je odkazem na konfiguraci prostředí, která určuje, jak se má aplikace připojit ke službě Event Hubs. Může zadat:

Pokud je nakonfigurovaná hodnota přesná shoda pro jedno nastavení i shodu předpony pro jiná nastavení, použije se přesná shoda.

Connection string

Tuto připojovací řetězec získáte kliknutím na tlačítko informace o Připojení pro obor názvů, nikoli samotným centrem událostí. Připojovací řetězec musí být pro obor názvů služby Event Hubs, nikoli pro samotné centrum událostí.

Při použití pro triggery musí mít připojovací řetězec k aktivaci funkce alespoň oprávnění ke čtení. Při použití pro výstupní vazby musí mít připojovací řetězec oprávnění k odesílání zpráv do datového proudu událostí.

Tato připojovací řetězec by měla být uložena v nastavení aplikace s názvem, který connection odpovídá hodnotě určené vlastností konfigurace vazby.

Připojení založená na identitách

Pokud používáte rozšíření verze 5.x nebo vyšší, můžete místo použití připojovací řetězec s tajným kódem použít aplikaci identitu Microsoft Entra. K tomu byste definovali nastavení pod běžnou předponou, která se mapuje na connection vlastnost v konfiguraci triggeru a vazby.

V tomto režimu rozšíření vyžaduje následující vlastnosti:

Vlastnost Šablona proměnné prostředí Popis Příklad hodnoty
Plně kvalifikovaný obor názvů <CONNECTION_NAME_PREFIX>__fullyQualifiedNamespace Plně kvalifikovaný obor názvů služby Event Hubs. myeventhubns.servicebus.windows.net

Pro přizpůsobení připojení je možné nastavit další vlastnosti. Viz Běžné vlastnosti pro připojení založená na identitě.

Poznámka:

Při použití konfigurace Aplikace Azure nebo služby Key Vault k poskytnutí nastavení pro připojení spravované identity by názvy nastavení měly používat platný oddělovač klíčů, například : místo /__ názvu, aby se názvy správně přeložily.

Například <CONNECTION_NAME_PREFIX>:fullyQualifiedNamespace.

Při hostovaní ve službě Azure Functions používají připojení založená na identitách spravovanou identitu. Identita přiřazená systémem se používá ve výchozím nastavení, i když je možné zadat identitu přiřazenou uživatelem s vlastnostmi a clientID vlastnostmicredential. Všimněte si, že konfigurace identity přiřazené uživatelem s ID prostředku se nepodporuje . Při spuštění v jiných kontextech, jako je místní vývoj, se místo toho použije vaše identita vývojáře, i když je možné ji přizpůsobit. Viz Místní vývoj s připojeními založenými na identitách.

Udělení oprávnění identitě

Jakákoli identita, kterou používáte, musí mít oprávnění k provedení zamýšlených akcí. U většiny služeb Azure to znamená, že potřebujete přiřadit roli v Azure RBAC pomocí předdefinovaných nebo vlastních rolí, které tato oprávnění poskytují.

Důležité

Cílová služba může zpřístupnit některá oprávnění, která nejsou nutná pro všechny kontexty. Pokud je to možné, dodržujte zásadu nejnižšího oprávnění a udělte identitě pouze požadovaná oprávnění. Pokud například aplikace potřebuje jen číst ze zdroje dat, použijte roli, která má oprávnění jen ke čtení. Přiřazení role, která také umožňuje zápis do této služby, by bylo nevhodné, protože by to bylo nadměrné oprávnění pro operaci čtení. Podobně byste chtěli zajistit, aby přiřazení role bylo vymezeno pouze nad prostředky, které je potřeba číst.

Budete muset vytvořit přiřazení role, které poskytuje přístup k centru událostí za běhu. Oborem přiřazení role může být obor názvů služby Event Hubs nebo samotné centrum událostí. Role správy, jako je vlastník , nestačí. Následující tabulka ukazuje předdefinované role, které se doporučují při použití rozšíření Event Hubs v normálním provozu. Vaše aplikace může vyžadovat další oprávnění na základě kódu, který napíšete.

Typ vazby Příklad předdefinovaných rolí
Trigger Příjemce dat služby Azure Event Hubs, vlastník dat služby Azure Event Hubs
Výstupní vazba Odesílatel dat služby Azure Event Hubs

nastavení host.json

Soubor host.json obsahuje nastavení, která řídí chování triggeru služby Event Hubs. Podrobnosti o dostupných nastaveních najdete v části nastavení host.json .

Další kroky