Google API BatchRequest：已建立的连接被主机中的软件中止 [英] Google API BatchRequest: An established connection was aborted by the software in your host machine

问题描述

我正在测试InsertAllRequest的Google BatchRequest（C＃）。一旦一个批次达到超过60个请求（约30,000个bigquery table rows />总共10,880,366个Http ContentLength），异常情况如下。当我关闭防火墙时也是如此。我在网上找到的解决方案，例如关闭http keep-alive，在这种情况下不起作用，因为我没有控制API如何使用HttpClient。

  System.Net.Http.HttpRequestException：将内容复制到流时出错。 ---> System.IO.IOException：无法从传输连接读取数据：建立的连接被主机中的软件中止。 ---> System.Net.Sockets.SocketException：建立的连接被主机中的软件中止
在System.Net.Sockets.Socket.BeginReceive（Byte []缓冲区，Int32偏移量，Int32大小，SocketFlags socketFlags，AsyncCallback回调，对象状态）
在System.Net.Sockets.NetworkStream.BeginRead（Byte []缓冲区，Int32偏移量，Int32大小，AsyncCallback回调，对象状态）
 ---内部异常堆栈跟踪结束--- System.Net.GZipWrapperStream.EndRead（IAsyncResult asyncResult）上的
 $ System.Net.Http.HttpClientHandler.WebExceptionWrapperStream.EndRead（IAsyncResult asyncResult）上的
 $ System.Net.Http中的
。 StreamToStreamCopy.StartRead（）
 ---内部异常堆栈跟踪结束--- Microsoft.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess（任务任务）
 Microsoft.Runtime.CompilerServices 
。 TaskAwaiter.HandleNonSuccess（任务任务）
在Microsoft.Runtime.CompilerServices.TaskAwaiter.ValidateEnd（任务任务）
在Microsoft.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult（）
 at Google.Apis.Requests.BatchRequest。< ExecuteAsync> d__3.MoveNext（）in c：\ApiaryDotnet\default\Src \GoogleApis\Apis\Requests\BatchRequest.cs：第175行
 ---从以前的位置抛出异常的堆栈跟踪结束--- 
在System.Runtime.CompilerServices.TaskAwaiter .ThrowForNonSuccess（任务任务）
在System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification（任务任务）
在System.Runtime.CompilerServices.TaskAwaiter.GetResult（）
在BigData.BigQuery.API .BigQueryHelper。< InsertBatchAsync>在c：\Users\fionazhao\Documents\BigDataCode\Framework\BigData\BigQuery\API\BigQueryHelper.cs中的d__1f.MoveNext（）：第197行
 ---抛出异常的前一个位置的堆栈跟踪结束--- System.Runtime.CompilerServices.TaskA中的
 $ System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification $（$任务任务）
 System.Runtime.CompilerServices.TaskAwaiter.GetResult（
  

以下是引发此异常的Google API代码：

  public async Task ExecuteAsync（CancellationToken cancellationToken）
 {
 if（Count< 1）
 return; 
 
 ConfigurableHttpClient httpClient = service.HttpClient; 
 
 var requests =从allRequests中的r 
选择r.ClientRequest; 
 HttpContent outerContent = await CreateOuterRequestContent（requests）.ConfigureAwait（false）; 
 var result = await httpClient.PostAsync（new Uri（batchUrl），outerContent，cancellationToken）
 .ConfigureAwait（false）; 
 
 result.EnsureSuccessStatusCode（）; 
 
 //获取边界分隔符。 
 const string boundaryKey =boundary =; 
 var fullContent = await result.Content.ReadAsStringAsync（）。ConfigureAwait（false）; 
 var contentType = result.Content.Headers.GetValues（Content-Type）。First（）; 
 var boundary = contentType.Substring（contentType.IndexOf（boundaryKey）+ boundaryKey.Length）; 
 
 int requestIndex = 0; 
 //当仍有内容需要读取时，解析当前的HTTP响应。 
 while（true）
 {
 cancellationToken.ThrowIfCancellationRequested（）; 
 
 var startIndex = fullContent.IndexOf（ - + boundary）; 
 if（startIndex == -1）
 {
 break; 
} 
 fullContent = fullContent.Substring（startIndex + boundary.Length + 2）; 
 var endIndex = fullContent.IndexOf（ - + boundary）; 
 if（endIndex == -1）
 {
 break; 
} 
 
 HttpResponseMessage responseMessage = ParseAsHttpResponse（fullContent.Substring（0，endIndex））; 
 
 if（responseMessage.IsSuccessStatusCode）
 {
 //解析当前内容对象。 
 var responseContent = await responseMessage.Content.ReadAsStringAsync（）。ConfigureAwait（false）; 
 var content = service.Serializer.Deserialize（responseContent，
 allRequests [requestIndex] .ResponseType）; 
 
 allRequests [requestIndex] .OnResponse（content，null，requestIndex，responseMessage）; 
} 
 else 
 {
 //解析当前响应中的错误。 
 var error = await service.DeserializeError（responseMessage）.ConfigureAwait（false）; 
 allRequests [requestIndex] .OnResponse（null，error，requestIndex，responseMessage）; 
} 
 
 requestIndex ++; 
 fullContent = fullContent.Substring（endIndex）; 
} 
} 
  

解决方案

略高于10MB，这是流式插入的请求大小限制。调整您的代码以适应这些限制：

最大行大小：1 MB

HTTP请求大小限制：10 MB

最大每秒行数：每个表格每秒100,000行。超过此数量将导致quota_exceeded错误。

每个请求的最大行数：500

每秒的最大字节数：每个表格每秒100 MB。超过此金额将导致quota_exceeded错误。

https：// cloud.google.com/bigquery/streaming-data-into-bigquery

I'm testing Google BatchRequest (C#) of InsertAllRequest. once a batch reaches more than 60 requests (~30,000 bigquery table rows / > 10,880,366 Http ContentLength in total), Exceptions happens as below. It's the same when I turned off my firewall. Solutions I found online such as turn http keep-alive off doesn't work in this case because I don't have controls on how the API uses the HttpClient.

    System.Net.Http.HttpRequestException: Error while copying content to a stream. ---> System.IO.IOException: Unable to read data from the transport connection: An established connection was aborted by the software in your host machine. ---> System.Net.Sockets.SocketException: An established connection was aborted by the software in your host machine
   at System.Net.Sockets.Socket.BeginReceive(Byte[] buffer, Int32 offset, Int32 size, SocketFlags socketFlags, AsyncCallback callback, Object state)
   at System.Net.Sockets.NetworkStream.BeginRead(Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   --- End of inner exception stack trace ---
   at System.Net.GZipWrapperStream.EndRead(IAsyncResult asyncResult)
   at System.Net.Http.HttpClientHandler.WebExceptionWrapperStream.EndRead(IAsyncResult asyncResult)
   at System.Net.Http.StreamToStreamCopy.StartRead()
   --- End of inner exception stack trace ---
   at Microsoft.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at Microsoft.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccess(Task task)
   at Microsoft.Runtime.CompilerServices.TaskAwaiter.ValidateEnd(Task task)
   at Microsoft.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
   at Google.Apis.Requests.BatchRequest.<ExecuteAsync>d__3.MoveNext() in c:\ApiaryDotnet\default\Src\GoogleApis\Apis\Requests\BatchRequest.cs:line 175
--- 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.GetResult()
   at BigData.BigQuery.API.BigQueryHelper.<InsertBatchAsync>d__1f.MoveNext() in c:\Users\fionazhao\Documents\BigDataCode\Framework\BigData\BigQuery\API\BigQueryHelper.cs:line 197
--- 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.GetResult()

Here's the Google's API code that throwed this exception:

 public async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        if (Count < 1)
            return;

        ConfigurableHttpClient httpClient = service.HttpClient;

        var requests = from r in allRequests
                       select r.ClientRequest;
        HttpContent outerContent = await CreateOuterRequestContent(requests).ConfigureAwait(false);
        var result = await httpClient.PostAsync(new Uri(batchUrl), outerContent, cancellationToken)
            .ConfigureAwait(false);

        result.EnsureSuccessStatusCode();

        // Get the boundary separator.
        const string boundaryKey = "boundary=";
        var fullContent = await result.Content.ReadAsStringAsync().ConfigureAwait(false);
        var contentType = result.Content.Headers.GetValues("Content-Type").First();
        var boundary = contentType.Substring(contentType.IndexOf(boundaryKey) + boundaryKey.Length);

        int requestIndex = 0;
        // While there is still content to read, parse the current HTTP response.
        while (true)
        {
            cancellationToken.ThrowIfCancellationRequested();

            var startIndex = fullContent.IndexOf("--" + boundary);
            if (startIndex == -1)
            {
                break;
            }
            fullContent = fullContent.Substring(startIndex + boundary.Length + 2);
            var endIndex = fullContent.IndexOf("--" + boundary);
            if (endIndex == -1)
            {
                break;
            }

            HttpResponseMessage responseMessage = ParseAsHttpResponse(fullContent.Substring(0, endIndex));

            if (responseMessage.IsSuccessStatusCode)
            {
                // Parse the current content object.
                var responseContent = await responseMessage.Content.ReadAsStringAsync().ConfigureAwait(false);
                var content = service.Serializer.Deserialize(responseContent,
                    allRequests[requestIndex].ResponseType);

                allRequests[requestIndex].OnResponse(content, null, requestIndex, responseMessage);
            }
            else
            {
                // Parse the error from the current response.
                var error = await service.DeserializeError(responseMessage).ConfigureAwait(false);
                allRequests[requestIndex].OnResponse(null, error, requestIndex, responseMessage);
            }

            requestIndex++;
            fullContent = fullContent.Substring(endIndex);
        }
    }

解决方案

That's just above 10MB which is the request size limit for streaming inserts. Adjust your code for these limits:

Maximum row size: 1 MB
HTTP request size limit: 10 MB
Maximum rows per second: 100,000 rows per second, per table. Exceeding this amount will cause quota_exceeded errors.
Maximum rows per request: 500
Maximum bytes per second: 100 MB per second, per table. Exceeding this amount will cause quota_exceeded errors.

https://cloud.google.com/bigquery/streaming-data-into-bigquery

这篇关于Google API BatchRequest：已建立的连接被主机中的软件中止的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！