Desencadenador de Azure Event Hubs para Azure Functions

Artículo
05/13/2021
Tiempo de lectura: 14 minutos

En este artículo se explica cómo usar el desencadenador de Azure Event Hubs para Azure Functions. Azure Functions admite los enlaces de salida y desencadenador para Event Hubs.

Para obtener información sobre los detalles de instalación y configuración, vea la información general.

Use el desencadenador de funciones para responder a un evento enviado a una secuencia de eventos del centro de eventos. Debe tener acceso de lectura al centro de eventos subyacente para configurar el desencadenador. Cuando esta función se desencadena, el mensaje que se pasa a la función se escribe como una cadena.

Ampliación

Cada instancia de una función de desencadenador de eventos está respaldada por una única instancia de EventProcessorHost. El desencadenador (con tecnología de Event Hubs) garantiza que solo una instancia de EventProcessorHost puede obtener una concesión en una partición determinada.

Por ejemplo, considere una instancia de Event Hubs con las siguientes características:

10 particiones.
1000 eventos distribuidos uniformemente en todas las particiones, con 100 mensajes en cada partición.

Cuando se habilita la función por primera vez, solo hay una instancia de la función. Vamos a llamar a esta instancia de función Function_0. Laq función Function_0 tiene una sola instancia de EventProcessorHost que contiene una concesión en las diez particiones. Esta instancia lee eventos de las particiones 0-9. A partir de este punto, se producirá una de las siguientes acciones:

No se necesitan nuevas instancias de función: Function_0 puede procesar los 1000 eventos antes de que la lógica de escalado de Azure Functions surta efecto. En este caso, Function_0 procesa los 1000 mensajes.
Se agrega una instancia de función adicional: Si la lógica de escalado de Azure Functions determina que Function_0 tiene más mensajes de los que puede procesar, se crea una nueva instancia de la aplicación de función (Function_1). Esta nueva función también tiene asociada una instancia de EventProcessorHost. Como la instancia de Event Hubs subyacente detecta que una nueva instancia de host está tratando de leer mensajes, efectúa un equilibrio de carga de las particiones entre las instancias de host. Por ejemplo, las particiones 0-4 pueden asignarse a Function_0 y las particiones 5-9, a Function_1.
Se agregan N instancias de función más: Si la lógica de escalado de Azure Functions determina que tanto Function_0 como Function_1 tienen más mensajes de los que pueden procesar, se crean más instancias de aplicaciones de función de Functions_N. Se van creando aplicaciones hasta llegar a un punto en el que N es mayor que el número de particiones de centro de eventos. En nuestro ejemplo, Event Hubs vuelve a equilibrar la carga de las particiones, en este caso, entre las instancias Function_0...Functions_9.

Cuando se produce el escalado, N instancias es un número mayor que el número de particiones del centro de eventos. Este patrón se usa para garantizar que va a haber instancias de EventProcessorHost disponibles para obtener bloqueos de las particiones a medida que estén disponibles en otras instancias. Solo se le cobra por los recursos usados cuando se ejecuta la instancia de la función. En otras palabras, no se le cobrará por este aprovisionamiento en exceso.

Cuando se completa la ejecución de todas las funciones con o sin errores, se agregan puntos de comprobación a la cuenta de almacenamiento asociada. Cuando estos puntos de conexión se agregan correctamente, los 1000 mensajes ya no se vuelven a recuperar.

En el ejemplo siguiente se muestra una función de C# que registra el cuerpo del mensaje del desencadenador del centro de eventos.

[FunctionName("EventHubTriggerCSharp")]
public static void Run([EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] string myEventHubMessage, ILogger log)
{
    log.LogInformation($"C# function triggered to process a message: {myEventHubMessage}");
}

Para acceder a los metadatos del evento en el código de función, cree un enlace a un objeto EventData (requiere el uso de una instrucción para Microsoft.Azure.EventHubs). También puede acceder a las mismas propiedades mediante el uso de expresiones de enlace en la firma del método. El ejemplo siguiente se muestran las dos maneras de obtener los mismos datos:

[FunctionName("EventHubTriggerCSharp")]
public static void Run(
    [EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] EventData myEventHubMessage,
    DateTime enqueuedTimeUtc,
    Int64 sequenceNumber,
    string offset,
    ILogger log)
{
    log.LogInformation($"Event: {Encoding.UTF8.GetString(myEventHubMessage.Body)}");
    // Metadata accessed by binding to EventData
    log.LogInformation($"EnqueuedTimeUtc={myEventHubMessage.SystemProperties.EnqueuedTimeUtc}");
    log.LogInformation($"SequenceNumber={myEventHubMessage.SystemProperties.SequenceNumber}");
    log.LogInformation($"Offset={myEventHubMessage.SystemProperties.Offset}");
    // Metadata accessed by using binding expressions in method parameters
    log.LogInformation($"EnqueuedTimeUtc={enqueuedTimeUtc}");
    log.LogInformation($"SequenceNumber={sequenceNumber}");
    log.LogInformation($"Offset={offset}");
}

Para recibir eventos en un lote, convierta string o EventData en una matriz.

Nota

Cuando se reciben en un lote, no se puede establecer un enlace a parámetros de método como en el ejemplo anterior con DateTime enqueuedTimeUtc y debe recibirlos desde cada objeto EventData.

[FunctionName("EventHubTriggerCSharp")]
public static void Run([EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] EventData[] eventHubMessages, ILogger log)
{
    foreach (var message in eventHubMessages)
    {
        log.LogInformation($"C# function triggered to process a message: {Encoding.UTF8.GetString(message.Body)}");
        log.LogInformation($"EnqueuedTimeUtc={message.SystemProperties.EnqueuedTimeUtc}");
    }
}

En el ejemplo siguiente se muestra un enlace de desencadenador de centro de eventos en un archivo function.json y una función de script de C# que usa el enlace. La función registra el cuerpo del mensaje del desencadenador de centro de eventos.

Los ejemplos siguientes muestran datos de enlace de Event Hubs en el archivo function.json.

Versión 2.x y posteriores

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

Versión 1.x

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "path": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

Este es el código de script de C#:

using System;

public static void Run(string myEventHubMessage, TraceWriter log)
{
    log.Info($"C# function triggered to process a message: {myEventHubMessage}");
}

#r "Microsoft.Azure.EventHubs"

using System.Text;
using System;
using Microsoft.ServiceBus.Messaging;
using Microsoft.Azure.EventHubs;

public static void Run(EventData myEventHubMessage,
    DateTime enqueuedTimeUtc,
    Int64 sequenceNumber,
    string offset,
    TraceWriter log)
{
    log.Info($"Event: {Encoding.UTF8.GetString(myEventHubMessage.Body)}");
    log.Info($"EnqueuedTimeUtc={myEventHubMessage.SystemProperties.EnqueuedTimeUtc}");
    log.Info($"SequenceNumber={myEventHubMessage.SystemProperties.SequenceNumber}");
    log.Info($"Offset={myEventHubMessage.SystemProperties.Offset}");

    // Metadata accessed by using binding expressions
    log.Info($"EnqueuedTimeUtc={enqueuedTimeUtc}");
    log.Info($"SequenceNumber={sequenceNumber}");
    log.Info($"Offset={offset}");
}

Para recibir eventos en un lote, convierta string o EventData en una matriz:

public static void Run(string[] eventHubMessages, TraceWriter log)
{
    foreach (var message in eventHubMessages)
    {
        log.Info($"C# function triggered to process a message: {message}");
    }
}

En el ejemplo siguiente se muestra un enlace de desencadenador de centro de eventos en un archivo function.json y una función de JavaScript que usa el enlace. La función lee los metadatos del evento y registra el mensaje.

Los ejemplos siguientes muestran datos de enlace de Event Hubs en el archivo function.json.

Versión 2.x y posteriores

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

Versión 1.x

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "path": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

Este es el código de JavaScript:

module.exports = function (context, myEventHubMessage) {
    context.log('Function triggered to process a message: ', myEventHubMessage);
    context.log('EnqueuedTimeUtc =', context.bindingData.enqueuedTimeUtc);
    context.log('SequenceNumber =', context.bindingData.sequenceNumber);
    context.log('Offset =', context.bindingData.offset);

    context.done();
};

Para recibir eventos en un lote, establezca cardinality en many en el archivo function.json, como se muestra en los ejemplos siguientes.

Versión 2.x y posteriores

{
  "type": "eventHubTrigger",
  "name": "eventHubMessages",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "cardinality": "many",
  "connection": "myEventHubReadConnectionAppSetting"
}

Versión 1.x

{
  "type": "eventHubTrigger",
  "name": "eventHubMessages",
  "direction": "in",
  "path": "MyEventHub",
  "cardinality": "many",
  "connection": "myEventHubReadConnectionAppSetting"
}

Este es el código de JavaScript:

module.exports = function (context, eventHubMessages) {
    context.log(`JavaScript eventhub trigger function called for message array ${eventHubMessages}`);

    eventHubMessages.forEach((message, index) => {
        context.log(`Processed message ${message}`);
        context.log(`EnqueuedTimeUtc = ${context.bindingData.enqueuedTimeUtcArray[index]}`);
        context.log(`SequenceNumber = ${context.bindingData.sequenceNumberArray[index]}`);
        context.log(`Offset = ${context.bindingData.offsetArray[index]}`);
    });

    context.done();
};

En el ejemplo siguiente se muestra un enlace de desencadenador de centro de eventos en un archivo function.json y una función de Python que usa el enlace. La función lee los metadatos del evento y registra el mensaje.

Los ejemplos siguientes muestran datos de enlace de Event Hubs en el archivo function.json.

{
  "type": "eventHubTrigger",
  "name": "event",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

Este es el código de Python:

import logging
import azure.functions as func


def main(event: func.EventHubEvent):
    logging.info(f'Function triggered to process a message: {event.get_body().decode()}')
    logging.info(f'  EnqueuedTimeUtc = {event.enqueued_time}')
    logging.info(f'  SequenceNumber = {event.sequence_number}')
    logging.info(f'  Offset = {event.offset}')

    # Metadata
    for key in event.metadata:
        logging.info(f'Metadata: {key} = {event.metadata[key]}')

En el ejemplo siguiente se muestra un enlace del centro de eventos que registra el cuerpo del mensaje del desencadenador de centro de eventos.

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

En la biblioteca en tiempo de ejecución de funciones de Java, utilice la anotación EventHubTrigger en los parámetros cuyo valor provendría del centro de eventos. Los parámetros con estas anotaciones hacen que la función se ejecuta cuando llega un evento. Esta anotación se puede usar con tipos nativos de Java, POJO o valores que aceptan valores NULL mediante Optional<T>.

Atributos y anotaciones

En las bibliotecas de clases de C#, use el atributo EventHubTriggerAttribute.

El constructor del atributo toma el nombre del centro de eventos, el nombre del grupo de consumidores y el nombre de una configuración de aplicación que contenga la cadena de conexión. Para obtener más información sobre estas configuraciones, vea la sección Configuración de desencadenador. Este es un ejemplo de atributo EventHubTriggerAttribute:

[FunctionName("EventHubTriggerCSharp")]
public static void Run([EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] string myEventHubMessage, ILogger log)
{
    ...
}

Para un ejemplo completo, consulte Desencadenador: ejemplo de C#.

Configuración

En la siguiente tabla se explican las propiedades de configuración de enlace que se definen en el archivo function.json y el atributo EventHubTrigger.

Propiedad de function.json	Propiedad de atributo	Descripción
type	N/D	Se debe establecer en `eventHubTrigger`. Esta propiedad se establece automáticamente cuando se crea el desencadenador en Azure Portal.
direction	N/D	Se debe establecer en `in`. Esta propiedad se establece automáticamente cuando se crea el desencadenador en Azure Portal.
name	N/D	Nombre de la variable que representa el elemento de evento en el código de la función.
path	EventHubName	Solo Functions 1.x. El nombre del centro de eventos. Cuando el nombre del centro de eventos también está presente en la cadena de conexión, ese valor reemplaza esta propiedad en tiempo de ejecución.
eventHubName	EventHubName	Functions 2.x y versiones posteriores. El nombre del centro de eventos. Cuando el nombre del centro de eventos también está presente en la cadena de conexión, ese valor reemplaza esta propiedad en tiempo de ejecución. Se puede hacer referencia a él desde la configuración de la aplicación `%eventHubName%`
consumerGroup	ConsumerGroup	Una propiedad opcional que establece el grupo de consumidores que se usará para suscribirse a los eventos del centro. Si se pasa por alto, se utilizará el grupo de consumidores `$Default`.
cardinalidad	N/D	Se utiliza para los lenguajes distintos de C#. Defínalo como `many` para permitir el procesamiento por lotes. Si se omite o se define como `one`, se pasa un único mensaje a la función. En C#, esta propiedad se asigna automáticamente siempre que el desencadenador tenga una matriz para el tipo.
connection	Connection	Nombre de una configuración de aplicación o de una colección de configuraciones de aplicación que especifica cómo conectarse a Event Hubs. Consulte Conexiones.

Cuando desarrolla localmente, la configuración de aplicación pasa al archivo local.settings.json.

Conexiones

La propiedad connection es una referencia a la configuración del entorno que especifica cómo se debe conectar la aplicación a Event Hubs. Puede especificar lo siguiente:

El nombre de una configuración de la aplicación que contiene una cadena de conexión.
El nombre de un prefijo compartido para varias configuraciones de la aplicación que juntas definen una conexión basada en identidad.

Si el valor configurado es tanto una coincidencia exacta de una única configuración como una coincidencia de prefijo de otras configuraciones, se usa la coincidencia exacta.

Cadena de conexión

Para obtener esta cadena de conexión, haga clic en el botón Información de conexión del espacio de nombres, no del propio centro de eventos. La cadena de conexión debe ser para un espacio de nombres de Event Hubs, no para el propio centro de eventos.

Cuando se usa para desencadenadores, la cadena de conexión debe tener al menos permisos de "lectura" para activar la función. Cuando se usa para los enlaces de salida, la cadena de conexión debe tener permisos de "envío" para enviar mensajes al flujo de eventos.

La cadena de conexión debe almacenarse en una configuración de la aplicación con un nombre que coincida con el valor especificado por la propiedad connection de la configuración de enlace.

Conexiones basadas en identidades

Si usa la versión 5.x o posterior de la extensión, en lugar de usar una cadena de conexión con un secreto, puede hacer que la aplicación use una identidad de Azure Active Directory. Para ello, definiría la configuración con un prefijo común que se asigne a la propiedad connection en la configuración de desencadenador y enlace.

En este modo, la extensión requiere las siguientes propiedades:

Propiedad	Plantilla de variable de entorno	Descripción	Valor de ejemplo
Espacio de nombres completo	`<CONNECTION_NAME_PREFIX>__fullyQualifiedNamespace`	Espacio de nombres completo de Event Hubs.	<event_hubs_namespace>.servicebus.windows.net

Se pueden establecer propiedades adicionales para personalizar la conexión. Consulte Propiedades comunes para conexiones basadas en identidades.

Cuando se hospeda en el servicio de Azure Functions, las conexiones basadas en identidades usan una identidad administrada. La identidad asignada por el sistema se usa de manera predeterminada, aunque se puede especificar una identidad asignada por el usuario con las propiedades credential y clientID. Cuando se ejecuta en otros contextos, como el de desarrollo local, se usa en su lugar la identidad del desarrollador, aunque se puede personalizar. Consulte Desarrollo local con conexiones basadas en identidades.

Concesión de permiso a la identidad

Cualquier identidad que se utilice debe tener permisos para realizar las acciones previstas. Deberá asignar un rol en Azure RBAC mediante roles integrados o personalizados que proporcionen esos permisos.

Importante

Es posible que el servicio de destino muestre algunos permisos que no son necesarios para todos los contextos. Siempre que sea posible, respete el principio de privilegios mínimos y conceda solo los privilegios necesarios a la identidad. Por ejemplo, si la aplicación solo necesita poder leer desde un origen de datos, use un rol que solo tenga permiso de lectura. Sería inadecuado asignar un rol que también permita escribir en ese servicio, ya que sería un permiso excesivo para una operación de lectura. De forma similar, le interesa asegurarse de que la asignación de roles esté limitada solo a los recursos que se deben leer.

Deberá crear una asignación de roles que proporcione acceso al centro de eventos en tiempo de ejecución. El ámbito de la asignación de roles debe ser para un espacio de nombres de Event Hubs, no para el propio centro de eventos. Los roles de administración, como Propietario, no son suficientes. En la tabla siguiente se muestran los roles integrados que se recomiendan al usar la extensión de Event Hubs con un funcionamiento normal. Puede que la aplicación precise permisos adicionales en función del código que escriba.

Tipo de enlace	Roles integrados de ejemplo
Desencadenador	[Receptor de los datos de Azure Event Hubs], [Propietario de los datos de Azure Event Hubs]
Enlace de salida	Emisor de datos de Azure Event Hubs

Uso

Valor predeterminado

Los siguientes tipos de parámetros se pueden usar en el centro de eventos de desencadenamiento:

string
byte[]
POCO
EventData - Las propiedades predeterminadas de EventData se proporcionan para el espacio de nombres Microsoft.Azure.EventHubs.

Tipos adicionales

Las aplicaciones que usan la versión 5.0.0 o posterior de la extensión del centro de eventos utilizan el tipo EventData en Azure.Messaging.EventHubs en lugar del que está en el espacio de nombres Microsoft.Azure.EventHubs. En esta versión se elimina la compatibilidad con el tipo Body heredado en favor de los siguientes tipos:

EventBody

Valor predeterminado

Los siguientes tipos de parámetros se pueden usar en el centro de eventos de desencadenamiento:

string
byte[]
POCO
EventData - Las propiedades predeterminadas de EventData se proporcionan para el espacio de nombres Microsoft.Azure.EventHubs.

Tipos adicionales

EventBody

Metadatos de evento

El desencadenador de Event Hubs proporciona varias propiedades de metadatos. Se pueden usar propiedades de metadatos como parte de expresiones de enlace en otros enlaces o como parámetros del código. Las propiedades proceden de la clase EventData.

Propiedad	Tipo	Descripción
`PartitionContext`	PartitionContext	Instancia de `PartitionContext`.
`EnqueuedTimeUtc`	`DateTime`	Hora de puesta en la cola en UTC.
`Offset`	`string`	El desplazamiento de los datos relacionados con el flujo de partición de Event Hubs. El desplazamiento es un marcador o identificador del flujo de Event Hubs. El identificador es único dentro de una partición del flujo de Event Hubs.
`PartitionKey`	`string`	La partición a la que se deben enviar los datos del evento.
`Properties`	`IDictionary<String,Object>`	Las propiedades de usuario de los datos del evento.
`SequenceNumber`	`Int64`	El número de secuencia de registro del evento.
`SystemProperties`	`IDictionary<String,Object>`	Las propiedades del sistema, incluidos los datos del evento.

Consulte los ejemplos de código que utilizan estas propiedades más arriba en este artículo.

configuración de host.json

El archivo host.json contiene la configuración que controla el comportamiento del desencadenador del centro de eventos. Consulte la sección de configuración de host.json para más información sobre las opciones de configuración disponibles.

Pasos siguientes

Escribir eventos en una secuencia de eventos (enlace de salida)

Desencadenador de Azure Event Hubs para Azure Functions

Valore su experiencia

Ampliación

Atributos y anotaciones

Configuración

Conexiones

Cadena de conexión

Conexiones basadas en identidades

Concesión de permiso a la identidad

Uso

Valor predeterminado

Tipos adicionales

Metadatos de evento

configuración de host.json

Pasos siguientes