node.js - 填充队列时 Azure Functions 中的超时

Question

我们有一个简单的 ETL 过程，用于将数据从 API 提取到文档数据库，我们希望使用函数来实现该过程。简而言之，该过程是获取约 16,500 行文件，从每行提取 ID(函数 1)，为每个 ID 构建 URL(函数 2)，使用 URL 调用 API(函数 3)，存储响应在文档 DB 中(功能 4)。我们正在使用队列进行函数间通信，并且在执行此操作时发现第一个函数存在超时问题。

函数 1 (index.js)

module.exports = function (context, odsDataFile) {
  context.log('JavaScript blob trigger function processed blob \n Name:', context.bindingData.odaDataFile, '\n Blob Size:', odsDataFile.length, 'Bytes');

  const odsCodes = [];

  odsDataFile.split('\n').map((line) => {
    const columns = line.split(',');

    if (columns[12] === 'A') {
      odsCodes.push({
        'odsCode': columns[0],
        'orgType': 'pharmacy',
      });
    }
  });

  context.bindings.odsCodes = odsCodes;
  context.log(`A total of: ${odsCodes.length} ods codes have been sent to the queue.`);

  context.done();
};

function.json

{
  "bindings": [
    {
      "type": "blobTrigger",
      "name": "odaDataFile",
      "path": "input-ods-data",
      "connection": "connecting-to-services_STORAGE",
      "direction": "in"
    },
    {
      "type": "queue",
      "name": "odsCodes",
      "queueName": "ods-org-codes",
      "connection": "connecting-to-services_STORAGE",
      "direction": "out"
    }
  ],
  "disabled": false
}

完整代码here

当 ID 的数量为 100 时，该函数工作正常，但当 ID 的数量为 10 或 1000 时，该函数就会超时。 ID 数组的构建在几毫秒内完成，并且函数完成，但是将项目添加到队列似乎需要很多分钟，并最终导致默认的 5 分钟超时。

令我惊讶的是，填充队列的简单操作似乎花费了如此长的时间，并且函数的超时似乎包括函数外部任务的时间(即队列填充)。这是可以预料的吗？有没有更高效的方法来做到这一点？

我们正在消耗(动态)计划下运行。

Answer 1

我在本地计算机上对此进行了一些测试，发现将消息插入队列需要大约 200 毫秒，这是预期的。因此，如果您有 17k 条消息要插入并且按顺序执行，则需要的时间为:

17,000 条消息 * 200 毫秒 = 3,400,000 毫秒或约 56 分钟

从云端运行时，延迟可能会更快一点，但您可以看到，当您插入这么多消息时，延迟会很快超过 5 分钟。

如果消息顺序并不重要，您可以并行插入消息。但有一些警告:

1. 你不能用 Node 来做到这一点——它必须是 C#。 Node 不会向您公开 IAsyncCollector 接口(interface)，因此它会在幕后完成所有操作。
2. 您无法并行插入所有内容，因为消耗计划限制一次最多有 250 个网络连接。

下面是一次批量插入 200 条的示例 - 包含 17k 条消息，在我的快速测试中这花费了不到一分钟的时间。

public static async Task Run(string myBlob, IAsyncCollector<string> odsCodes, TraceWriter log)
{
    List<Task> tasks = new List<Task>();
    string[] lines = myBlob.Split(Environment.NewLine.ToCharArray(), StringSplitOptions.RemoveEmptyEntries);

    int skip = 0;
    int take = 200;

    IEnumerable<string> batch = lines.Skip(skip).Take(take);

    while (batch.Count() > 0)
    {
        await AddBatch(batch, odsCodes);
        skip += take;
        batch = lines.Skip(skip).Take(take);
    }
}

public static async Task AddBatch(IEnumerable<string> lines, IAsyncCollector<string> odsCodes)
{
    List<Task> tasks = new List<Task>();    
    foreach (string line in lines)
    {
        tasks.Add(odsCodes.AddAsync(line));
    }
    await Task.WhenAll(tasks);
}