Azure Data Lake Store - 现有连接被远程主机强行关闭

Question

我使用 DataLakeStoreFileSystemManagementClient class 从 Data Lake Store 读取文件。我们用这样的代码为文件打开一个steam，逐字节读取并处理它。这是一个特殊情况，我们不能使用 U-SQL 进行数据处理。

m_adlsFileSystemClient = new DataLakeStoreFileSystemManagementClient(…);
return m_adlsFileSystemClient.FileSystem.OpenAsync(m_connection.AccountName, path);

读取和处理文件的过程最多可能需要 60 分钟。问题是：我经常收到“现有连接被远程主机强行关闭”。流读取过程异常。特别是当阅读需要 20 分钟或更长时间时。它不应该是超时，因为我使用正确的客户端超时设置创建了 DataLakeStoreFileSystemManagementClient。您可以在下面找到异常详细信息。异常看起来是随机的，很难预测你何时得到它。处理时间可以是第15分钟，也可以是第50分钟。

从Data Lake Store读取文件正常吗？对于 Data Lake Store 中的文件保持打开流的总时间是否有任何限制（或建议）？

异常：

   System.AggregateException: One or more errors occurred. -
--> System.IO.IOException: Unable to read data from the transport connection: An
existing connection was forcibly closed by the remote host. ---> System.Net.Soc
kets.SocketException: An existing connection was forcibly closed by the remote h
ost
   at System.Net.Sockets.Socket.Receive(Byte[] buffer, Int32 offset, Int32 size,
SocketFlags socketFlags)
   at System.Net.Sockets.NetworkStream.Read(Byte[] buffer, Int32 offset, Int32 s
ize)
   --- End of inner exception stack trace ---
   at System.Net.ConnectStream.Read(Byte[] buffer, Int32 offset, Int32 size)
   at System.Net.Http.HttpClientHandler.WebExceptionWrapperStream.Read(Byte[] bu
ffer, Int32 offset, Int32 count)
   at System.Net.Http.DelegatingStream.Read(Byte[] buffer, Int32 offset, Int32 c
ount)
   at DataLake.Timeout.Research.FileDownloader.CopyStream(Stream input, Stream o
utput) in C:\TFS-SED\Main\Platform\DataNode\DataLake\DataLake.Timeout.Research\F
ileDownloader.cs:line 107
   at DataLake.Timeout.Research.FileDownloader.<DownloadFileAsync>d__6.MoveNext(
) in C:\TFS-SED\Main\Platform\DataNode\DataLake\DataLake.Timeout.Research\FileDo
wnloader.cs:line 96

Answer 1

为避免此类问题，建议采取以下措施：

读取更小的、可重试的块。根据我的经验，我发现 4MB 块效果最好，性能最好。此外，通过以较小的增量读取，您可以将重试逻辑合并到在失败的情况下从相同的偏移量重试。

如果您不知道您的流有多大（例如，它正在在您阅读时由另一名工人附加），您可以检查对于负载中带有 RemoteException 的 400 错误 “BadOffsetException”。这将表明您已经开始超出文件末尾的偏移量。

const int MAX_BYTES_TO_READ = 4 * 1024 * 1024; //4MB
…
long offset = 0;
while(notDone)
{
try
{
                var myStream = client.Read(accountName, offset, MAX_BYTES_TO_READ)
                // do stuff with stream
}
catch(WebException ex)
{
                // read the web exception response
                If (response.contains(“BadOffsetException”))
                                noteDone = false;
}
}

Answer 2

谢谢阿米特，你的建议终于帮到我了。那是我的版本。该示例正在使用重试逻辑读取一批字节。我通过具有 4MB 缓冲区的 BufferedSteam 使用它。所以客户端可以一个对象一个对象地读取流，但是我们以 4MB 的批次请求服务。

while (!m_endOfFile)
{
    try
    {
        var inputStream = m_client.OpenReadFile(
            m_filePath,
            length: count, 
            offset: m_position);

            var memoryStream = new MemoryStream(count);
            inputStream.CopyTo(memoryStream);
            m_position += memoryStream.Length;
            result = memoryStream.ToArray();
            break;
    }
    catch (CloudException ex)
    {
        if (ex.Response.Content.ToString().Contains("Invalid offset value"))
        {
            m_endOfFile = true;
        }
        else
        {
            throw;
        }
        }
    catch (IOException)
    {
        repeats++;
        if (repeats >= RepeatCount)
        {
            throw;
        }
    }  
}

Azure Data Lake Store - 现有连接被远程主机强行关闭

Azure Data Lake Store - existing connection was forcibly closed by the remote host

azure-data-lake