1

我在由 Priam 管理的 Cassandra 集群上运行在 Amazon 上的 Java 应用程序。

我们使用 Amazon 的 Elastic Map/Reduce 服务，在某个时刻，当我运行 EMR 并尝试在 Cassandra 上插入一些数据时，我得到了一个异常：OperationTimeoutException。

这些是我在 Astyanax 上创建 Cassandra 池时传递的配置参数：

`ConnectionPoolConfigurationImpl conPool = new` `ConnectionPoolConfigurationImpl(getConecPoolName())`
    .setMaxConnsPerHost(20)
        .setSeeds("ec2-xx-xxx-xx-xx.compute-1.amazonaws.com")
    .setMaxOperationsPerConnection(100)                       .setMaxPendingConnectionsPerHost(20) 
    .setConnectionLimiterMaxPendingCount(20) 
    .setTimeoutWindow(10000) 
    .setConnectionLimiterWindowSize(1000) 
    .setMaxTimeoutCount(3) 
    .setConnectTimeout(5000) 
    .setMaxFailoverCount(-1) 
    .setLatencyAwareBadnessThreshold(20)
        .setLatencyAwareUpdateInterval(1000)
    .setLatencyAwareResetInterval(10000) 
        .setLatencyAwareWindowSize(100) 
    .setLatencyAwareSentinelCompare(100f) 


AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder()
        .forCluster("clusterName")
        .forKeyspace("keyspaceName")
    .withAstyanaxConfiguration(
           new AstyanaxConfigurationImpl().setDiscoveryType(NodeDiscoveryType.NONE))
    .withConnectionPoolConfiguration(conPool)
    .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
    .buildKeyspace(ThriftFamilyFactory.getInstance());

完整的堆栈跟踪：

ERROR com.s1mbi0se.dg.input.service.InputService (main): EXCEPTION:OperationTimeoutException: [host=ec2-xx-xxx-xx-xx.compute-1.amazonaws.com(10.100.6.242):9160, latency=10004(10004), attempts=1]TimedOutException()

com.netflix.astyanax.connectionpool.exceptions.OperationTimeoutException: OperationTimeoutException: [host=ec2-54-224-65-18.compute-1.amazonaws.com(10.100.6.242):9160, latency=10004(10004), attempts=1]TimedOutException()
    at com.netflix.astyanax.thrift.ThriftConverter.ToConnectionPoolException(ThriftConverter.java:171)
    at com.netflix.astyanax.thrift.AbstractOperationImpl.execute(AbstractOperationImpl.java:61)
    at com.netflix.astyanax.thrift.ThriftColumnFamilyQueryImpl$1$2.execute(ThriftColumnFamilyQueryImpl.java:206)
    at com.netflix.astyanax.thrift.ThriftColumnFamilyQueryImpl$1$2.execute(ThriftColumnFamilyQueryImpl.java:198)
    at com.netflix.astyanax.thrift.ThriftSyncConnectionFactoryImpl$ThriftConnection.execute(ThriftSyncConnectionFactoryImpl.java:151)
    at com.netflix.astyanax.connectionpool.impl.AbstractExecuteWithFailoverImpl.tryOperation(AbstractExecuteWithFailoverImpl.java:69)
    at com.netflix.astyanax.connectionpool.impl.AbstractHostPartitionConnectionPool.executeWithFailover(AbstractHostPartitionConnectionPool.java:253)
    at com.netflix.astyanax.thrift.ThriftColumnFamilyQueryImpl$1.execute(ThriftColumnFamilyQueryImpl.java:196)
    at com.s1mbi0se.dg.input.service.InputService.searchUserByKey(InputService.java:833)
    at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:771)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:375)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1132)
    at org.apache.hadoop.mapred.Child.main(Child.java:249)
Caused by: TimedOutException()
    at org.apache.cassandra.thrift.Cassandra$get_slice_result.read(Cassandra.java:7874)
    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:78)
    at org.apache.cassandra.thrift.Cassandra$Client.recv_get_slice(Cassandra.java:594)
    at org.apache.cassandra.thrift.Cassandra$Client.get_slice(Cassandra.java:578)
    at com.netflix.astyanax.thrift.ThriftColumnFamilyQueryImpl$1$2.internalExecute(ThriftColumnFamilyQueryImpl.java:211)
    at com.netflix.astyanax.thrift.ThriftColumnFamilyQueryImpl$1$2.internalExecute(ThriftColumnFamilyQueryImpl.java:198)
    at com.netflix.astyanax.thrift.AbstractOperationImpl.execute(AbstractOperationImpl.java:56)

所以我不知道解决这个问题的方向是什么，因为问题可能出在 Astyanax 池配置、EC2 机器配置（内存增加？）、Priam 配置或 AWS 上 Cassandra 或 EMR 服务所需的其他配置在我的代码中......任何提示？

遵循堆栈跟踪：

INFO org.apache.hadoop.mapred.TaskLogsTruncater (main): Initializing logs' truncater with mapRetainSize=-1 and reduceRetainSize=-1
WARN org.apache.hadoop.mapred.Child (main): Error running child
java.lang.RuntimeException: InvalidRequestException(why:Start key's token sorts after end token)
    at org.apache.cassandra.hadoop.ColumnFamilyRecordReader$WideRowIterator.maybeInit(ColumnFamilyRecordReader.java:453)
    at org.apache.cassandra.hadoop.ColumnFamilyRecordReader$WideRowIterator.computeNext(ColumnFamilyRecordReader.java:459)
    at org.apache.cassandra.hadoop.ColumnFamilyRecordReader$WideRowIterator.computeNext(ColumnFamilyRecordReader.java:406)
    at com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:143)
    at com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:138)
    at org.apache.cassandra.hadoop.ColumnFamilyRecordReader.getProgress(ColumnFamilyRecordReader.java:103)
    at org.apache.hadoop.mapred.MapTask$NewTrackingRecordReader.getProgress(MapTask.java:522)
    at org.apache.hadoop.mapred.MapTask$NewTrackingRecordReader.nextKeyValue(MapTask.java:547)
    at org.apache.hadoop.mapreduce.MapContext.nextKeyValue(MapContext.java:67)
    at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:143)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:771)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:375)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1132)
    at org.apache.hadoop.mapred.Child.main(Child.java:249)
Caused by: InvalidRequestException(why:Start key's token sorts after end token)
    at org.apache.cassandra.thrift.Cassandra$get_paged_slice_result.read(Cassandra.java:14168)
    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:78)
    at org.apache.cassandra.thrift.Cassandra$Client.recv_get_paged_slice(Cassandra.java:769)
    at org.apache.cassandra.thrift.Cassandra$Client.get_paged_slice(Cassandra.java:753)
    at org.apache.cassandra.hadoop.ColumnFamilyRecordReader$WideRowIterator.maybeInit(ColumnFamilyRecordReader.java:438)
    ... 
INFO org.apache.hadoop.mapred.Task (main): Runnning cleanup for the task
4

2 回答 2

1

我们解决了这个问题（院长，我在 Cassandra 用户组上回答了这个问题，但我会再次说明我们在这里为解决问题所做的工作）

  • 首先我们将 Cassandra 更新为 1.2.3 版本
  • 更新 Cassandra 后，启动了一个新异常“没有可借用的主机”，我们发现命令“ConnectionPoolConfigurationImpl(...).setConnectTimeout(-1)”是原因...
  • 我们把 .setConnectTimeout(2000)
  • 我们增加了 Astyanax 池中的其他值，我们的应用程序终于工作了......

基本上我认为我们最初的问题是亚马逊延迟太高，所以我们改变了我们的池配置并且一切正常......

感谢所有人的帮助（主要是院长）！

在我们在 Amazon 上运行的实际池配置下方：

new ConnectionPoolConfigurationImpl(getConecPoolName())
.setMaxConnsPerHost(CONNECTION_POOL_SIZE_PER_HOST)
.setSeeds(getIpSeeds())
    .setMaxOperationsPerConnection(10000) 
    .setMaxPendingConnectionsPerHost(20) 
    .setConnectionLimiterMaxPendingCount(20)    
        .setTimeoutWindow(10000) 
    .setConnectionLimiterWindowSize(2000)
    .setMaxTimeoutCount(3) 
    .setConnectTimeout(100) 
    .setConnectTimeout(2000)
    .setMaxFailoverCount(-1) 
    .setLatencyAwareBadnessThreshold(20)
    .setLatencyAwareUpdateInterval(1000) // 10000
    .setLatencyAwareResetInterval(10000) // 60000
    .setLatencyAwareWindowSize(100) // 100
    .setLatencyAwareSentinelCompare(100f)                      .setSocketTimeout(30000)
    .setMaxTimeoutWhenExhausted(10000)
    .setInitConnsPerHost(10)
        ;

AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder().forCluster(clusterName).forKeyspace(keyspaceName)
                .withAstyanaxConfiguration(new AstyanaxConfigurationImpl().setDiscoveryType(NodeDiscoveryType.NONE).setConnectionPoolType(ConnectionPoolType.ROUND_ROBIN).setDiscoveryDelayInSeconds(10000)
        .setDiscoveryDelayInSeconds(10000))
        .withConnectionPoolConfiguration(conPool)
            .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
        .buildKeyspace(ThriftFamilyFactory.getInstance());
于 2013-04-18T13:16:30.587 回答
0

那么，如果将超时设置为 -1 会发生什么？就个人而言，我会深入研究 astyanax 代码并尝试找出如何禁用超时。再次运行你的东西，它应该会继续运行，尽管如果你遇到超时，你的集群当然可能会受到重创......我假设你对此没问题。

编辑（后期编辑后）：拍摄，我忘了问你使用的是哪个版本的 cassandra。我正在查看此代码，但第 346 行是您的第 438 行（您正在使用 Widerow 迭代器，这很可能意味着行扫描（即行的片段）。

http://grepcode.com/file/repo1.maven.org/maven2/org.apache.cassandra/cassandra-all/1.2.2/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java#ColumnFamilyRecordReader.0split

我们至少可以在其中看到这是一个键范围，但由于行可能太宽而分页（对于内存来说，对于不太宽的行还有另一个迭代器）。我相信您是正确的，您不能使用两种分区器类型。要获得有关此的更多信息，我强烈建议修改 ColumnFamilyRecordReader.java 以记录 ColumnFamilySplit（它上面有一个 toString）。您可以在初始化方法中记录它，也可以记录 JobRange（它也有一个 toString ）。

IE。

logger.warn("my split range="+split+" job's total range="+jobRange);

您的版本将与此代码有很多相似之处。你在哪个版本？

除了拆分之外，我还将记录 KeySlice，以防万一，如果我没记错的话，任何一个都可能导致该错误。让我知道您使用的版本并添加一些日志以获取有关您情况的更多信息。（他们的东西通常很容易开箱即用地编译，没有问题）。

院长

于 2013-04-16T17:02:15.883 回答