java - 运行大量数据时与 platfOrm 的连接重置

Question

我有一个 hadoop 进程，它连接到reduce 部分中的cassandra 键空间。数据由playORM保存。 发生的情况是:我在同一台机器上运行这个 hadoop 进程和 cassandra，因此 playORM 仅连接到本地主机上的 cassandra。当我处理少量数据时，该过程运行完全正常，但当我处理大量数据(在本例中仅 500 000 条记录)时，我收到以下异常。 我想知道这是否可能是 astyanax 池配置中的问题(这是由 playORM 完成的，所以我不知道如何更改这些设置)，或者是否可能是 playORM 本身甚至我的 Cassandra 配置中的问题。现在一切都在单个主机上运行，​​我认为当我们配置集群时情况可能会变得更糟，因为许多 hadoop 机器将连接到许多 cassandra 机器。

有什么可能出问题的提示吗？

CF=[tablename=Localization] persist rowkey=1bd9b46a-5b66-41ae-9756-dd91f44194ea
CF=User index persist(cf=[tablename=User])=[rowkey=/User/id] (table found, colmeta not found)
CF=[tablename=User] persist rowkey=1bd9b46a-5b66-41ae-9756-dd91f44194ea
java.lang.RuntimeException: com.netflix.astyanax.connectionpool.exceptions.ConnectionAbortedException: ConnectionAbortedException: [host=localhost(127.0.0.1):9160, latency=611(611), attempts=1] org.apache.thrift.t
ransport.TTransportException: java.net.SocketException: Connection reset
        at com.alvazan.orm.layer9z.spi.db.cassandra.CassandraSession.sendChanges(CassandraSession.java:110)
        at com.alvazan.orm.logging.NoSqlRawLogger.sendChanges(NoSqlRawLogger.java:50)
        at com.alvazan.orm.layer5.nosql.cache.NoSqlWriteCacheImpl.flush(NoSqlWriteCacheImpl.java:125)
        at com.alvazan.orm.layer5.nosql.cache.NoSqlReadCacheImpl.flush(NoSqlReadCacheImpl.java:178)
        at com.alvazan.orm.layer0.base.BaseEntityManagerImpl.flush(BaseEntityManagerImpl.java:182)
        at com.s1mbi0se.dmp.da.dao.UserDao.insertOrUpdateUser(UserDao.java:24)
        at com.s1mbi0se.dmp.da.dao.UserDao.insertOrUpdateUserLocalization(UserDao.java:75)
        at com.s1mbi0se.dmp.da.service.DataAccessService.insertLocalizationForUser(DataAccessService.java:44)
        at com.s1mbi0se.dmp.module.LocalizationModule.persistData(LocalizationModule.java:218)
        at com.s1mbi0se.dmp.processor.mapred.SelectorReducer.reduce(SelectorReducer.java:60)
        at com.s1mbi0se.dmp.processor.mapred.SelectorReducer.reduce(SelectorReducer.java:1)
        at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
        at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:649)
        at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:417)
        at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:260)
Caused by: com.netflix.astyanax.connectionpool.exceptions.ConnectionAbortedException: ConnectionAbortedException: [host=localhost(127.0.0.1):9160, latency=611(611), attempts=1] org.apache.thrift.transport.TTranspo
rtException: java.net.SocketException: Connection reset
        at com.netflix.astyanax.thrift.ThriftConverter.ToConnectionPoolException(ThriftConverter.java:193)
        at com.netflix.astyanax.thrift.AbstractOperationImpl.execute(AbstractOperationImpl.java:60)
        at com.netflix.astyanax.thrift.AbstractOperationImpl.execute(AbstractOperationImpl.java:27)
        at com.netflix.astyanax.thrift.ThriftSyncConnectionFactoryImpl$1.execute(ThriftSyncConnectionFactoryImpl.java:131)
        at com.netflix.astyanax.connectionpool.impl.AbstractExecuteWithFailoverImpl.tryOperation(AbstractExecuteWithFailoverImpl.java:52)
        at com.netflix.astyanax.connectionpool.impl.AbstractHostPartitionConnectionPool.executeWithFailover(AbstractHostPartitionConnectionPool.java:229)
        at com.netflix.astyanax.thrift.ThriftKeyspaceImpl.executeOperation(ThriftKeyspaceImpl.java:455)
        at com.netflix.astyanax.thrift.ThriftKeyspaceImpl.access$400(ThriftKeyspaceImpl.java:62)
        at com.netflix.astyanax.thrift.ThriftKeyspaceImpl$1.execute(ThriftKeyspaceImpl.java:115)
        at com.alvazan.orm.layer9z.spi.db.cassandra.CassandraSession.sendChangesImpl(CassandraSession.java:131)
        at com.alvazan.orm.layer9z.spi.db.cassandra.CassandraSession.sendChanges(CassandraSession.java:108)
        ... 14 more
Caused by: org.apache.thrift.transport.TTransportException: java.net.SocketException: Connection reset
        at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:129)
        at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
        at org.apache.thrift.transport.TFramedTransport.readFrame(TFramedTransport.java:129)
        at org.apache.thrift.transport.TFramedTransport.read(TFramedTransport.java:101)
        at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
        at org.apache.thrift.protocol.TBinaryProtocol.readAll(TBinaryProtocol.java:378)
        at org.apache.thrift.protocol.TBinaryProtocol.readI32(TBinaryProtocol.java:297)
        at org.apache.thrift.protocol.TBinaryProtocol.readMessageBegin(TBinaryProtocol.java:204)
        at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:69)
        at org.apache.cassandra.thrift.Cassandra$Client.recv_batch_mutate(Cassandra.java:913)
        at org.apache.cassandra.thrift.Cassandra$Client.batch_mutate(Cassandra.java:899)
        at com.netflix.astyanax.thrift.ThriftKeyspaceImpl$1$1.internalExecute(ThriftKeyspaceImpl.java:121)
        at com.netflix.astyanax.thrift.ThriftKeyspaceImpl$1$1.internalExecute(ThriftKeyspaceImpl.java:118)
        at com.netflix.astyanax.thrift.AbstractOperationImpl.execute(AbstractOperationImpl.java:55)
        ... 23 more
Caused by: java.net.SocketException: Connection reset
        at java.net.SocketInputStream.read(Unknown Source)
        at java.net.SocketInputStream.read(Unknown Source)
        at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:127)
        ... 36 more

Answer 1

注意:我想我也遇到过这个问题，并提高了 astyanax 中的超时或连接池大小，它消失了，所以也尝试一下(尽管连接重置通常是远端服务器的错误......即 cassandra) .

连接重置通常是因为另一端(cassandra)关闭了您的连接。为了 100% 确定，如果您使用wireshark，您应该看到哪一端正在关闭套接字。

请小心您在这篇文章中读到的内容...

java.net.SocketException: Connection reset

但基本上，我在 mina、netty 等出现之前就在 sourceforge 上编写了 Channelmanager。大多数情况下，当另一端正确关闭套接字时，您会得到-1......即。他们需要发送一些数据包。如果它们消失，可能会导致连接重置等异常。

我建议摆弄 astyanax 连接池。看看wireshark，然后google一下tcp拆卸是如何发生的，看看cassandra是否没有正确拆卸它。

如果您使用的是 Linux，请尝试 netstat -anp | grep {pid} 这样你就可以看到你的客户端进程正在使用的端口，并在wireshark中查找这些端口上的数据包。另外，做一个测试以确保 astyanax 正确地保持其池，这意味着在此过程中运行几次 netstat 命令，以确保 astyanax 不会创建套接字，然后删除它们并再次创建它们(就好像它删除了一个并且然后你写入它，你可能会得到上面的错误)

java nio 的东西在幕后从来都不是完全可靠的......直到今天，我仍然有单元测试来展示不同操作系统上 nio 库中的错误。

出于好奇，当我注意到您正在执行写入操作并且读取操作基本上无法获取写入成功与否的状态时，您还向管道中冲洗了多少量。

在接下来的几个月中，我们希望有一个通用的 Map/Reduce 来为 Map/Reduce 代码提供实际的实体。我们终于找到了一位新开发人员，并向他发送了报价，该开发人员很快就会加入我们，帮助我们分担工作量。

另一篇值得阅读的好文章是这个

http://kb.realvnc.com/questions/75/I%27m+receiving+the+error+%22Connection+reset+by+peer+%2810054%29%22.+

wireshark 确实可以告诉您 tcp 层发生的情况的详细信息。我一直想调查更多细节，是 astyanax 还是 cassandra 的错，但没有时间。

院长