azure-durable-functions - Azure Durable Functions 大消息支持问题

Question

我在将大量对象从事件函数返回到协调器函数时遇到问题。
我有一个下载 180 MB 文件并解析它的函数。该文件将生成一个包含超过 962K 条目的对象列表。每个对象有大约 70 个属性，但只有大约 20% 被填充。
当我运行该函数时，代码成功下载文件并将其解析到列表中，但是当返回列表时，会引发异常并显示以下信息:

异常:“执行函数时出现异常:#######”
- 来源:“System.Private.CoreLib”

内部异常:“在函数返回后处理参数 $return 时出错。”
- 来源:“Microsoft.Azure.WebJobs.Host”

内部/内部异常:“抛出了‘System.OutOfMemoryException’类型的异常。”
- 来源:“System.Private.CoreLib”

最后一个嵌套异常将 NewtonsoftJson 包列为进行调用的包，该包生成报告的内存不足错误。我在最后包含了此异常的完整堆栈跟踪。

我知道我可以序列化对象列表并将它们存储在 Azure blob 条目中，然后在需要处理它的下一个函数中再次提取它，但我认为持久函数背后的想法是避免所有这些并且保持更精简的工作流程？此外，我的设计基于“大消息支持 #26”github 帖子，该帖子指出，如果大小超过队列消息限制，持久函数扩展将自动将函数有效负载存储在 blob 中(请参阅:https://github.com/Azure/azure-functions-durable-extension/issues/26)。

有什么我需要做的吗？
代码非常简单:

[FunctionName("GetDataFromSource")]
public static IEnumerable<DataDetail> GetDataFromSource([ActivityTrigger]ISource source, ILogger logger)
{
    try
    {
        string importSettings = Environment.GetEnvironmentVariable(source.SettingsKey);
        if (string.IsNullOrWhiteSpace(importSettings))
        {
            logger.LogError($"No settings key information found for the {source.SourceId} data source");                    }
        else
        {
            List<DataDetail> _Data = source.GetVinData().Distinct().ToList();
            return vinData;
         }
     }
     catch (Exception ex)
     {
         logger.LogCritical($"Error processing the {source.SourceId} Vin data source. *** Exception: {ex}");
     }

      return new List<DataDetail>();
}

这是最内部异常的堆栈跟踪:

at System.Text.StringBuilder.ExpandByABlock(Int32 minBlockCharCount)
   at System.Text.StringBuilder.Append(Char value, Int32 repeatCount)
   at System.Text.StringBuilder.Append(Char value)
   at System.IO.StringWriter.Write(Char value)
   at Newtonsoft.Json.JsonTextWriter.WritePropertyName(String name, Boolean escape)
   at Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeObject(JsonWriter writer, Object value, JsonObjectContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
   at Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, Object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
   at Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeList(JsonWriter writer, IEnumerable values, JsonArrayContract contract, JsonProperty member, JsonContainerContract collectionContract, JsonProperty containerProperty)
   at Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.SerializeValue(JsonWriter writer, Object value, JsonContract valueContract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerProperty)
   at Newtonsoft.Json.Serialization.JsonSerializerInternalWriter.Serialize(JsonWriter jsonWriter, Object value, Type objectType)
   at Newtonsoft.Json.JsonSerializer.SerializeInternal(JsonWriter jsonWriter, Object value, Type objectType)
   at DurableTask.Core.Serializing.JsonDataConverter.Serialize(Object value, Boolean formatted)
   at Microsoft.Azure.WebJobs.Extensions.DurableTask.MessagePayloadDataConverter.Serialize(Object value, Int32 maxSizeInKB) in C:\projects\azure-functions-durable-extension\src\WebJobs.Extensions.DurableTask\MessagePayloadDataConverter.cs:line 55
   at Microsoft.Azure.WebJobs.Extensions.DurableTask.MessagePayloadDataConverter.Serialize(Object value) in C:\projects\azure-functions-durable-extension\src\WebJobs.Extensions.DurableTask\MessagePayloadDataConverter.cs:line 43
   at Microsoft.Azure.WebJobs.DurableActivityContext.SetOutput(Object output) in C:\projects\azure-functions-durable-extension\src\WebJobs.Extensions.DurableTask\DurableActivityContext.cs:line 136
   at Microsoft.Azure.WebJobs.Extensions.DurableTask.ActivityTriggerAttributeBindingProvider.ActivityTriggerBinding.ActivityTriggerReturnValueBinder.SetValueAsync(Object value, CancellationToken cancellationToken) in C:\projects\azure-functions-durable-extension\src\WebJobs.Extensions.DurableTask\Bindings\ActivityTriggerAttributeBindingProvider.cs:line 213
   at Microsoft.Azure.WebJobs.Host.Executors.FunctionExecutor.ParameterHelper.ProcessOutputParameters(CancellationToken cancellationToken) in C:\projects\azure-webjobs-sdk-rqm4t\src\Microsoft.Azure.WebJobs.Host\Executors\FunctionExecutor.cs:line 972

Answer 1

我在使用 Durable Functions 时遇到了类似的问题。
有几个解决方案/解决方法:
正如您所说，您可以将函数负载存储在 blob 存储中，并在需要时检索它们。这有效，但会影响性能，并且可能需要一段时间才能检索，具体取决于您的文件有多大。
另一种选择是批量调用。我不完全确定你的 GetVinData()方法可以，但您可以修改它，以便一次只能检索 50,000(o x 个)项目。您的协调器可以多次调用您的事件函数并在协调器中建立您的列表。

[FunctionName(nameof(OrchestratorAsync))]
public async Task OrchestratorAsync([OrchestrationTrigger] IDurableOrchestrationContext context) 
{
    var dataDetailList = new List<DataDetail>();
    var batches = BuildBatchesHere();

    foreach (var batch in batches) 
    {
        dataDetailList.AddRange(
             await context.CallActivityAsync<List<DataDetail>>(
                  nameof(GetDataFromSource), batch);
    }
    // Do whatever you need with dataDetailList
}