Table of Contents
Introducción
Cuando se trabaja con Azure Functions, un escenario común es la importación de un archivo para su procesamiento. En nuestro caso concreto, nos encontramos con un archivo Excel con miles de líneas que deben procesarse, comprobando la exactitud de los valores mediante múltiples reglas y fuentes de datos, seguido del envío de notificaciones. Al principio, con un número reducido de líneas, el tiempo de procesamiento era aceptable. Sin embargo, a medida que aumentaba el tamaño del archivo Excel, hasta alcanzar decenas de miles de líneas, el tiempo de procesamiento se hacía insoportable. Para superar este reto, exploraremos las capacidades de Durable Functions y aprovecharemos su potencia para abordar este problema de forma eficaz.
Patrón Fan-out/fan-in
El patrón «fan-out/fan-in» es una técnica utilizada para paralelizar y consolidar la ejecución de múltiples tareas. Durante la fase fan-out, las tareas se inician en paralelo para realizar diferentes actividades de forma concurrente. A continuación, durante la fase fan-in, la función espera a que se completen todas las tareas paralelas y combina sus resultados. Este patrón permite una ejecución paralela eficiente, reduce el tiempo de procesamiento y permite la consolidación de datos o resultados obtenidos de tareas paralelas. Se trata de un enfoque escalable que maximiza la utilización de los recursos y mejora la eficiencia general de Durable Azure Functions.
En nuestro escenario particular, utilizaremos el patrón fan out/fan in para abordar la tarea de procesamiento de datos, que requiere mucho tiempo. En lugar de gestionar todo el proceso en una única función, distribuiremos y paralelizaremos la carga de trabajo en varias funciones. Gracias a este enfoque, el tiempo de procesamiento se reducirá significativamente, especialmente a medida que aumente el número de entradas en el archivo. Mediante el patrón fan out/fan in, podemos dividir y conquistar eficientemente la tarea de procesamiento de datos, logrando un mejor rendimiento y eficiencia general.
Paso 1: Cargar un archivo
El paso inicial en el flujo de trabajo de procesamiento de datos implica la carga de un archivo. Esto se consigue activando la función principal a través del BlobTrigger, que monitoriza el almacenamiento especificado en busca de nuevas adiciones de archivos. Para iniciar el patrón Fan-out/Fan-in, la función principal invocará a la función Orchestrator. La razón de tener funciones separadas es que cada función sólo puede tener un único disparador. Para pasar múltiples parámetros a la función Orchestrator, se puede utilizar un objeto serializable. Vale la pena señalar que los parámetros complejos, como flujos o entidades intrincadas, pueden encontrar problemas durante la transmisión. Además, asegúrese de que las Funciones están configuradas a 64 bits para manejar grandes volúmenes de datos con eficacia. La función Orchestrator se activa utilizando el método IDurableOrchestrationClient.StartNewAsync.
[FunctionName("ImportExcel")]
public async Task ImportExcel(
[DurableClient] IDurableOrchestrationClient orchestrationClient,
[BlobTrigger("storage/{name}")]
Stream file,
string name,
IDictionary<string, string> metaData,
ILogger log)
{
var data = await ParseExcel(file);
var dataToBeProcessed = new DataToBeProcessed { Data = data, FileName = name, Email = metaData["email"]};
var id = await orchestrationClient.StartNewAsync("DataOrchestrator", dataToBeProcessed);
log.LogInformation("End ImportExcel. Orchestrator instance id: {Id}", id);
}Paso 2: Llamar al Orquestador
El siguiente paso consiste en iniciar la función Orchestrator, que se encarga de gestionar el estado del flujo de trabajo de procesamiento de datos. Dentro de la función Orchestrator, el gran conjunto de datos se divide en trozos más pequeños, en nuestro ejemplo con un tamaño de 1000 entradas por trozo. Para procesar estos trozos en paralelo, se llama a una función de actividad auxiliar utilizando el enlace activityTrigger. La función ProcessData se invoca de la siguiente manera:
ProcessData([ActivityTrigger] PartialData inputs...)
los datos provocaría un aumento del tiempo de procesamiento a medida que aumentara el tamaño del archivo. Al aprovechar el patrón fan-out/fan-in, varias funciones se ejecutan simultáneamente en varias máquinas virtuales, lo que mejora el rendimiento. Los ayudantes de actividad se invocan utilizando el método context.CallActivityAsync.
La función Orchestrator esperará a que se completen todas las tareas paralelas utilizando la sentencia await Task.WhenAll(parallelTasks). Una vez recopiladas todas las respuestas, se llama a la función de actividad helper final. En este caso, el propósito de la actividad helper final es notificar al usuario que los datos han sido procesados con éxito.
El orquestador no puede contener ninguna llamada await fuera del contexto de orquestación. Cualquier llamada asíncrona debe procesarse en las funciones de la actividad.
[FunctionName("DataOrchestrator")]
public async Task DataOrchestrator([OrchestrationTrigger] IDurableOrchestrationContext context, ILogger log)
{
var input = context.GetInput<DataToBeProcessed>();
var data = input.Data;
var email = input.MetaData["email"];
var parallelTasks = new List<Task<List<ResponseDto>>>();
var split2 = data.Models.Chunk(1000);
foreach (var partialList in split2)
{
var task = context.CallActivityAsync<List<ResponseDto>>("ProcessData", partialList);
parallelTasks.Add(task);
}
await Task.WhenAll(parallelTasks);
var resultVendors = new List<ResponseDto>();
foreach (var task in parallelTasks)
{
resultPslVendors.AddRange(task.Result.response);
}
var resultLists = new ResultLists
{
ResultVendors = resultVendors,
FileName = metaData.FileName,
Email = email
};
await context.CallActivityAsync("SendNotifications", resultLists);
}Paso 3: Función de actividad: Tratamiento de los datos
El núcleo de la tarea de procesamiento de datos reside en la función de actividad. Mientras que en los enfoques tradicionales habría una única función que manejaría todo el conjunto de datos, en el patrón fan-out/fan-in, múltiples funciones trabajan en paralelo para procesar los datos. Esto permite un procesamiento eficiente y rápido, especialmente cuando se trata de grandes conjuntos de datos.
Cada función paralela realiza su tarea asignada de forma independiente, procesando un subconjunto de los datos. Al distribuir la carga de trabajo entre varias funciones, el tiempo total de procesamiento se reduce significativamente. Esta capacidad de procesamiento paralelo es una ventaja clave del patrón fan-out/fan-in, ya que aprovecha la escalabilidad y los recursos de la infraestructura subyacente de Azure.
La función de actividad desempeña un papel crucial en el procesamiento eficiente de los datos mediante la ejecución de tareas en paralelo y garantizando una utilización óptima de los recursos. Este enfoque no solo mejora el rendimiento, sino que también permite aprovechar mejor la potencia de cálculo disponible, lo que se traduce en un procesamiento de datos más rápido y eficiente.
[FunctionName("ProcessData")]
public async Task<List<ResponseDto>> ProcessData([ActivityTrigger] PartialData inputs, ILogger log)
{
return await ProcessExcelFile(inputs.Models, inputs.Token, inputs.Email);
}
Paso 4: Función Actividad: Envío de notificaciones
Una vez completadas todas las tareas paralelas del patrón fan-out/fan-in, la función orquestadora toma el control y ordena a otra función de actividad que envíe notificaciones. Este último paso garantiza que se notifique a los usuarios la finalización de la tarea de procesamiento de datos.
La función de actividad responsable del envío de notificaciones es activada por el orquestador mediante el enlace activityTrigger. Toma los datos y parámetros necesarios para generar y enviar las notificaciones apropiadas a los destinatarios previstos.
Al separar el proceso de envío de notificaciones en su propia función de actividad, la arquitectura general del sistema resulta más modular y fácil de mantener. También permite una extensibilidad más sencilla, ya que las modificaciones o mejoras del mecanismo de notificación pueden realizarse de forma independiente sin afectar a los demás componentes del proceso.
[FunctionName("SendNotifications")]
public async Task SendNotifications([ActivityTrigger] ResultLists inputs, ILogger log)
{
await SendNotifications(inputs.ResultPslVendors, inputs.Email);
}Comprobación de registros
Si examina los registros, podrá realizar un seguimiento del inicio de la función orquestadora, el número de actividades necesarias (en este caso, cinco) y el momento en que se enviaron las notificaciones.
2023-06-30T13:18:48Z [Information] Executed 'DataOrchestrator' (Succeeded, Id=892088e1-815d-422d-99c2-555187e2a20d, Duration=2170ms)
2023-06-30T13:27:18Z [Information] Executing 'DataOrchestrator' (Reason='(null)', Id=5cd3b96a-a36c-49ac-9acf-378f89ea548b)
2023-06-30T13:27:18Z [Information] af0e4d7735e54208bb65274993f2fc85: Function 'DataOrchestrator (Orchestrator)' started. IsReplay: False. Input: (22171712 bytes). State: Started. SequenceNumber: 79. TaskEventId: -1
2023-06-30T13:27:18Z [Information] af0e4d7735e54208bb65274993f2fc85: Function 'ProcessData (Activity)' scheduled. Reason: DataOrchestrator. IsReplay: False. State: Scheduled. SequenceNumber: 80.
2023-06-30T13:27:18Z [Information] af0e4d7735e54208bb65274993f2fc85: Function 'ProcessData (Activity)' scheduled. Reason: DataOrchestrator. IsReplay: False. State: Scheduled. SequenceNumber: 81.
2023-06-30T13:27:18Z [Information] af0e4d7735e54208bb65274993f2fc85: Function 'ProcessData (Activity)' scheduled. Reason: DataOrchestrator. IsReplay: False. State: Scheduled. SequenceNumber: 82.
2023-06-30T13:27:18Z [Information] af0e4d7735e54208bb65274993f2fc85: Function 'ProcessData (Activity)' scheduled. Reason: DataOrchestrator. IsReplay: False. State: Scheduled. SequenceNumber: 83.
2023-06-30T13:27:18Z [Information] af0e4d7735e54208bb65274993f2fc85: Function 'ProcessData (Activity)' scheduled. Reason: DataOrchestrator. IsReplay: False. State: Scheduled. SequenceNumber: 84.
2023-06-30T13:28:46Z [Information] Executed ‘DataOrchestrator' (Succeeded, Id=27198a53-60c2-41e1-9c8e-8a20abd8180d, Duration=1092ms)
Conclusión
La implementación del patrón fan-out/fan-in en las funciones Azure duraderas supuso una mejora significativa del tiempo de procesamiento. Al dividir los datos en varias tareas paralelas y aprovechar la potencia de la informática distribuida, se optimizó el proceso, que requería mucho tiempo. Esta transición de una única función a múltiples funciones que trabajan en paralelo demuestra la eficacia de este patrón para mejorar el rendimiento y la eficiencia generales.
