我使用datastax驱动程序将cassandra用作apache flink的某些数据流的接收器:我在执行应用程序时遇到问题,在运行时引发了一个关于队列的错误,该错误在几秒钟后变满。我发现默认值是256,这对于我的负载来说可能太低了,所以我使用pooligoptions设置maxrequestsperconnection来提高它,如下所示:http://docs.datastax.com/en/developer/java-driver/3.1/manual/pooling/.
不幸的是,对于以下代码,我在启动时出现以下错误:
The implementation of the ClusterBuilder is not serializable.
The object probably contains or references non serializable fields.
我的代码:
PoolingOptions poolingOptions = new PoolingOptions();
poolingOptions
.setMaxRequestsPerConnection(HostDistance.LOCAL, 32768)
.setMaxRequestsPerConnection(HostDistance.REMOTE, 10000);
ClusterBuilder cassandraBuilder = new ClusterBuilder() {
@Override
public Cluster buildCluster(Cluster.Builder builder) {
return builder.addContactPoint(CASSANDRA_ADDRESS)
.withPort(CASSANDRA_PORT)
.withPoolingOptions(poolingOptions)
.build();
}
};
sinkBuilderNormalStream
.setQuery("INSERT INTO keyspace_local.values_by_sensors_users"
+ " (user, sensor, timestamp, rdf_stream, observed_value, value)"
+ " VALUES (?, ?, ?, ?, ?, ?);")
.setClusterBuilder(cassandraBuilder)
.build();
我该怎么处理?
1条答案
按热度按时间hk8txs481#
您必须在clusterbuilder#buildcluster中定义pooligoptions。