java - 运行大量数据时使用 platfOrm 重置连接

Question

我有一个连接到 reduce 部分中的 cassandra 键空间的 hadoop 进程。数据由 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 并用谷歌搜索 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 的错，但没有时间。

院长