如何在Azure表存储中使用partitionkey加快查询速度

Question

我们如何提高查询速度？

在执行以下查询的范围内，我们大约有100个消费者1-2 minutes。这些运行中的每个运行都代表一个消耗函数的运行。

        TableQuery<T> treanslationsQuery = new TableQuery<T>()
         .Where(
          TableQuery.CombineFilters(
            TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, sourceDestinationPartitionKey)
           , TableOperators.Or,
            TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, anySourceDestinationPartitionKey)
          )
         );

该查询将产生大约5000个结果。

完整代码：

    public static async Task<IEnumerable<T>> ExecuteQueryAsync<T>(this CloudTable table, TableQuery<T> query) where T : ITableEntity, new()
    {
        var items = new List<T>();
        TableContinuationToken token = null;

        do
        {
            TableQuerySegment<T> seg = await table.ExecuteQuerySegmentedAsync(query, token);
            token = seg.ContinuationToken;
            items.AddRange(seg);
        } while (token != null);

        return items;
    }

    public static IEnumerable<Translation> Get<T>(string sourceParty, string destinationParty, string wildcardSourceParty, string tableName) where T : ITableEntity, new()
    {
        var acc = CloudStorageAccount.Parse(Environment.GetEnvironmentVariable("conn"));
        var tableClient = acc.CreateCloudTableClient();
        var table = tableClient.GetTableReference(Environment.GetEnvironmentVariable("TableCache"));
        var sourceDestinationPartitionKey = $"{sourceParty.ToLowerTrim()}-{destinationParty.ToLowerTrim()}";
        var anySourceDestinationPartitionKey = $"{wildcardSourceParty}-{destinationParty.ToLowerTrim()}";

        TableQuery<T> treanslationsQuery = new TableQuery<T>()
         .Where(
          TableQuery.CombineFilters(
            TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, sourceDestinationPartitionKey)
           , TableOperators.Or,
            TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, anySourceDestinationPartitionKey)
          )
         );

        var over1000Results = table.ExecuteQueryAsync(treanslationsQuery).Result.Cast<Translation>();
        return over1000Results.Where(x => x.expireAt > DateTime.Now)
                           .Where(x => x.effectiveAt < DateTime.Now);
    }

在这些执行期间，当有100个使用者时，如您所见，请求将聚集并形成峰值：

在这些高峰期间，请求通常需要1分钟以上的时间：

我们如何提高查询速度？

Answer 1

您可以考虑以下 3 件事：

1 . 首先，摆脱Where对查询结果执行的子句。最好在查询中尽可能多地包含子句（如果表上的任何索引也包含它们，那就更好了）。现在，您可以按如下方式更改查询：

var translationsQuery = new TableQuery<T>()
.Where(TableQuery.CombineFilters(
TableQuery.CombineFilters(
    TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, sourceDestinationPartitionKey),
    TableOperators.Or,
    TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, anySourceDestinationPartitionKey)
    ),
TableOperators.And,
TableQuery.CombineFilters(
    TableQuery.GenerateFilterConditionForDate("affectiveAt", QueryComparisons.LessThan, DateTime.Now),
    TableOperators.And,
    TableQuery.GenerateFilterConditionForDate("expireAt", QueryComparisons.GreaterThan, DateTime.Now))
));

因为您有大量数据要检索，所以最好并行运行查询。所以，你应该用我基于Stephen Toub Parallel.While编写的方法替换do whileloop insideExecuteQueryAsync方法；这样，它将减少查询执行时间。这是一个不错的选择，因为您可以在调用此方法时删除它，但是它有一点限制，我将在这部分代码之后讨论它：Parallel.ForEachResult

public static IEnumerable<T> ExecuteQueryAsync<T>(this CloudTable table, TableQuery<T> query) where T : ITableEntity, new()
{
    var items = new List<T>();
    TableContinuationToken token = null;

    Parallel.ForEach(new InfinitePartitioner(), (ignored, loopState) =>
    {
        TableQuerySegment<T> seg = table.ExecuteQuerySegmented(query, token);
        token = seg.ContinuationToken;
        items.AddRange(seg);

        if (token == null) // It's better to change this constraint by looking at https://www.vivien-chevallier.com/Articles/executing-an-async-query-with-azure-table-storage-and-retrieve-all-the-results-in-a-single-operation
            loopState.Stop();
    });

    return items;
}

然后你可以在你的Get方法中调用它：

return table.ExecuteQueryAsync(translationsQuery).Cast<Translation>();

正如您所看到的，方法本身不是异步的（您应该更改它的名称）并且Parallel.ForEach与传入异步方法不兼容。这就是我ExecuteQuerySegmented改用的原因。但是，使其具有更好的性能和使用异步方法的所有优点，可以代替上述ForEach带环ActionBlock的方法，数据流或ParallelForEachAsync从扩展方法AsyncEnumerator NuGet包。

2 .执行独立的并行查询，然后合并结果是一个不错的选择，即使其性能提升最多为 10%。这让您有时间能够找到性能最佳的友好查询。但是，永远不要忘记在其中包含所有约束，并测试两种方式以了解哪一种更适合您的问题。

3 . 我不确定这是否是一个好建议，但请执行并查看结果。如MSDN 中所述：

表服务按如下方式强制执行服务器超时：

• 查询操作：在超时间隔期间，查询最多可以执行 5 秒。如果查询未在 5 秒间隔内完成，则响应包含用于检索后续请求中剩余项目的继续令牌。有关更多信息，请参阅查询超时和分页。

• 插入、更新和删除操作：最大超时间隔为 30 秒。30 秒也是所有插入、更新和删除操作的默认间隔。

如果您指定的超时时间小于服务的默认超时时间，则将使用您的超时间隔。

所以你可以玩超时并检查是否有任何性能改进。

更新 06-30-2021

感谢 @WouterVanRanst 仔细研究了上面的代码片段，我决定更新它并使用另一个Parallel.ForEach方法重载，使循环单线程并防止TableContinuationToken. 你可以找到一个示例分区局部变量的说明这里在MSDN上。这是ExecuteQueryAsync<T>方法的新外观：

public static IEnumerable<T> ExecuteQueryAsync<T>(this CloudTable table, TableQuery<T> query) where T : ITableEntity, new()
{
    TableContinuationToken token = null;
    var items = new List<T>();

    Parallel.ForEach(new InfinitePartitioner(), () =>
    {
        return null as TableQuerySegment<T>;
    }, (ignored, loopState, segment) =>
    {
        segment = table.ExecuteQuerySegmented(query, token) as TableQuerySegment<T>;
        
        token = segment.ContinuationToken;

        if (token == null)
            loopState.Stop();

        return segment;
    },
    (seg) => items.AddRange(seg)
    );

    return items;
}

注意： 当然，您可以改进上面的代码或找到更好的方法来防止竞争条件，但这是一个很简单的方法。我会很高兴听到你的想法。

Answer 2

  var over1000Results = table.ExecuteQueryAsync(treanslationsQuery).Result.Cast<Translation>();
        return over1000Results.Where(x => x.expireAt > DateTime.Now)
                           .Where(x => x.effectiveAt < DateTime.Now);

这是问题之一，您正在运行查询，然后使用这些“wheres”从内存中过滤它。将过滤器移至查询运行之前，这应该会有很大帮助。

其次，您必须提供一些从数据库检索的行数限制