在 Google Cloud Spanner 中批量插入期间出现瞬时故障异常

Question

我目前正在为一个项目测试 Google Cloud Spanner 作为 MySQL 的替代品，因为我预计随着时间的推移，行数会增长到数亿。数据库需要非常快速地响应并在几秒钟内返回查询结果，所以我想我可以试试 Spanner。

我尝试将示例数据批量加载到我的 Spanner 数据库中，但我不断收到此错误：

未处理的异常：System.AggregateException：发生一个或多个错误。 ---> Google.Cloud.Spanner.Data.SpannerException：操作已中止。 ---> Grpc.Core.RpcException: Status(StatusCode=Aborted, Detail="Aborted due to transient fault")

我在这里使用了一个稍微修改过的代码：https://github.com/GoogleCloudPlatform/dotnet-docs-samples/blob/master/spanner/api/Program.cs

这里是InsertSampleData中的代码修改：

public static object InsertSampleData(string projectId,
        string instanceId, string databaseId)
    {
        // I get about 100k rows here
        List<Data> data = get_data();

        // how many runs I need to make if I split the data by 100 rows
        int rows = 100;
        double cnt = (double)data.Count / rows;
        cnt = Math.Ceiling(cnt);

        // process the data part by part
        for (int i = 0; i < cnt; i++)
        {
            // returns part of the data based on offset and amount
            List<Data> data_part = get_part(data, i, rows);

            var response = InsertTradesAsync(
            projectId, instanceId, databaseId, data_part);
            s_logger.Info("Waiting for operation to complete...");
            response.Wait();
            s_logger.Info($"Operation status: {response.Status}");
        }

        return ExitCode.Success;
    }

InsertTradesAsync与repo中的相同（当然参数列表除外）。

当我运行代码时，我总是收到以下错误：

Unhandled Exception: System.AggregateException: One or more errors occurred. ---> Google.Cloud.Spanner.Data.SpannerException: The operation was aborted. ---> Grpc.Core.RpcException: Status(StatusCode=Aborted, Detail="Aborted due to transient fault")
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.ValidateEnd(Task task)
   at Google.Api.Gax.Grpc.ApiCallRetryExtensions.<>c__DisplayClass0_0`2.<<WithRetry>b__0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.V1.Internal.ExecuteHelper.<WithSessionChecking>d__0`1.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.V1.TransactionPool.<RunFinalMethodAsync>d__9`1.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.Data.SpannerTransaction.<<CommitAsync>b__29_0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.Data.ExecuteHelper.<WithErrorTranslationAndProfiling>d__2`1.MoveNext()
   --- End of inner exception stack trace ---
   at Google.Cloud.Spanner.Data.ExecuteHelper.<WithErrorTranslationAndProfiling>d__2`1.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.Data.EphemeralTransaction.<>c__DisplayClass2_0.<<ExecuteMutationsAsync>b__1>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.Data.ExecuteHelper.<WithErrorTranslationAndProfiling>d__2`1.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Google.Cloud.Spanner.Data.SpannerCommand.<ExecuteMutationsAsync>d__49.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter`1.GetResult()
   at GoogleCloudSamples.Spanner.Program.<InsertTradesAsync>d__25.MoveNext() in c:\Users\user\Documents\Dev\spanner\api\Program.cs:line 1298
   --- End of inner exception stack trace ---
   at System.Threading.Tasks.Task.ThrowIfExceptional(Boolean includeTaskCanceledExceptions)
   at System.Threading.Tasks.Task.Wait(Int32 millisecondsTimeout, CancellationToken cancellationToken)
   at System.Threading.Tasks.Task.Wait()
   at GoogleCloudSamples.Spanner.Program.InsertSampleData(String projectId, String instanceId, String databaseId) in c:\Users\user\Documents\Dev\spanner\api\Program.cs:line 1585
   at GoogleCloudSamples.Spanner.Program.<>c__DisplayClass45_0.<Main>b__2(InsertSampleDataOptions opts) in c:\Users\user\Documents\Dev\spanner\api\Program.cs:line 1932
   at CommandLine.ParserResultExtensions.MapResult[T1,T2,T3,T4,T5,T6,T7,T8,T9,T10,T11,T12,T13,T14,T15,T16,TResult](ParserResult`1 result, Func`2 parsedFunc1, Func`2 parsedFunc2, Func`2 parsedFunc3, Func`2 parsedFunc4, Func`2 parsedFunc5, Func`2 parsedFunc6, Func`2 parsedFunc7, Func`2 parsedFunc8, Func`2 parsedFunc9, Func`2 parsedFunc10, Func`2 parsedFunc11, Func`2 parsedFunc12, Func`2 parsedFunc13, Func`2 parsedFunc14, Func`2 parsedFunc15, Func`2 parsedFunc16, Func`2 notParsedFunc)
   at GoogleCloudSamples.Spanner.Program.Main(String[] args) in c:\Users\user\Documents\Dev\spanner\api\Program.cs:line 1910

我认为这可能与限制和配额有关 (https://cloud.google.com/spanner/quotas)，但是在将不同数量的行插入我的表后引发异常（它似乎是随机的，有时会在 68 次运行 100 行后发生每个，然后是 28x100、52x100 等）。该表有 30 列，PK 由 2 列（没有索引）组成，我正在按 100 部分处理数据，所以我认为我没有达到极限。

如果我将 cmd.CommandTimeout 设置为一个非常高的数字，我会得到更高的插入行（大约 400x100） - 我假设客户端库重用连接？但是，我找不到任何有关 C# 库的信息或池化信息。即使插入了更多行，错误仍然存​​在。

非常感谢任何帮助。

谢谢！

Answer 1

谷歌开发者在这里。 我有一些建议可能会对您有所帮助。

首先，您应该使用单个事务并尽可能多地向其中添加写入。看起来你一次做100个？但你可以做得更多。您可以在这里看到可以批量写入的限制（我相信是 20,000）：

https://github.com/GoogleCloudPlatform/dotnet-docs-samples/blob/master/spanner/api/Program.cs#L1242

第二个建议直接解决了您的问题。由于 spanner 的工作方式，您需要在事务周围使用重试，如下面的示例所示：

https://github.com/GoogleCloudPlatform/dotnet-docs-samples/blob/master/spanner/api/Program.cs#L1259

（在此处下载瞬态故障应用程序 nuget）： https://www.nuget.org/packages/EnterpriseLibrary.TransientFaultHandling/

您需要这个，因为 spanner 可能偶尔会遇到死锁，这会迫使您重新运行整个事务。我们在 Exception 上提供了一个扩展方法“IsTransientSpannerFault”，以便更轻松地构建您的重试策略，如下所示：

    internal class CustomTransientErrorDetectionStrategy
        : ITransientErrorDetectionStrategy
    {
        public bool IsTransient(Exception ex) =>
            ex.IsTransientSpannerFault();
    }

希望这会有所帮助！

编辑：我刚刚注意到您没有等待对 InsertTradesAsync 的异步调用的结果。您可能希望至少在最后做一个 Task.WaitAll 。请注意，每批 100 次写入可能与前一批并行在后台运行。这很可能是导致出现强制重试的死锁几率增加的原因。

如果您这样做是为了提高性能，Spanner ADO.NET 库将在后台池中打开多个连接，这些连接等于 SpannerConnection.SpannerOptions.MaximumGrpcChannels。因此，您将看到一定程度的收益。您可以增加此值来调整性能。

问候