通过 azure 函数以同步方式同时处理多个文件

Question

我有一个 Azure 函数，它从 blob 容器中提取名称和文件 URL，然后将此信息发送到另一个函数以处理这些文件（解压缩并将它们保存在 Datalake 中）

对于 blob 的提取：

string storageConnectionString = @"myconnstring";
CloudStorageAccount storageAccount = CloudStorageAccount.Parse(storageConnectionString);
CloudBlobClient blobClient = storageAccount.CreateCloudBlobClient();
CloudBlobContainer container = blobClient.GetContainerReference("Container");

IEnumerable<IListBlobItem> blobs = new IListBlobItem[0];

foreach (IListBlobItem blobItem in container.ListBlobs())
     {                
        if (blobItem is CloudBlobDirectory)
           {
              CloudBlobDirectory directory = (CloudBlobDirectory)blobItem;
              blobs = directory.ListBlobs(true);                    
           }

      }

 await ProcessBlobs(blobs); 

以及处理螺栓的功能：

public static async Task ProcessBlobs(IEnumerable<IListBlobItem> blobs)
    {

        var tasks = blobs.Select(currentblob =>
        {
            string FileUrl = currentblob.Uri.ToString();
            string FileName = currentblob.Uri.Segments.Last();
            //string content = "{ \"fileUri\": \""+ currentblob.Uri.ToString()+ "\" , \"fileName\": \""+ currentblob.Uri.Segments.Last()+"\"}";

            var values = new Dictionary<string, string>
                {
                    { "fileUri", currentblob.Uri.ToString() },
                    { "fileName", currentblob.Uri.Segments.Last() }
                };
            var content = new FormUrlEncodedContent(values);
            string baseURL = @"https://<afu>.azurewebsites.net/api/process_zip_files_by_http_trigger?code=45"; ;
            //string urlToInvoke = string.Format("{0}&name={1}", baseURL, FileUrl, FileName);


            return RunAsync(baseURL, content);
        });
        await Task.WhenAll(tasks);
    }

    public static async Task RunAsync(string i_URL, FormUrlEncodedContent content)
    {
        var response = await client.PostAsync(i_URL, content);
        var responseString = await response.Content.ReadAsStringAsync();
        log.info(responseString);
    }

函数RunAsync异步处理文件。

我现在的问题是： 通常是否可以并行处理 blob，但以同步方式？你有更好更简单的想法来实现我的目标吗？

Answer 1

这是Durable Functions 的最佳用例之一。具体来说，Fan-Out/Fan-In Pattern。

你总共需要 4 个函数

1. GetBlobList Activity Function
   这是您获取要处理的 blob 列表的地方。您只会得到 blob 列表，而不是实际的 blob。

2. ProcessBlob Activity Function
   此函数采用 blob 路径，获取 blob 并对其进行处理。

3. Orchestrator Function
   这是调用 GetBlobList 函数、循环返回的 blob 路径列表并为每个 blob 路径调用 ProcessBlob 函数的函数。

4. 启动函数 (Client Function)
   此函数只是触发编排的运行。这通常是 HTTP 触发的函数。

如果您是 Durable Functions 的新手，最好通过quickstart doc 了解所需的不同类型的函数。