标签: astyanax

我们有一个创造性地命名为"bootstrap"的过程,它为环境中的给定rev软件设置我们的Cassandra集群(Dev1,Dev2,QA,...,PROD).此引导程序创建/更新键空间和列族以及在非prod中填充初始数据.

我们正在使用Astyanax,但我们可以使用Hector进行自举.

鉴于另一个团队已决定每个环境都有自己的数据中心名称.而且,当我们从两个数据中心转到更多数据中心时,我希望这可以在prod中工作.鉴于我们将使用PropertyFileSnitch:

我如何询问Cassandra集群的布局？(没有炮击nodetool ring)

具体来说,我需要知道数据中心的名称,以便在使用NetworkTopologyStrategy时可以使用正确的策略选项设置创建或更新密钥空间.我们希望每个数据中心有3个副本.有些envs有一个和几个有两个,最终生产会有更多.

是否有CQL或Thrift调用可以提供有关群集布局的信息？

我已经查看了各种文档集中的几个TOC,并用Google搜索了一下.我想在挖掘nodetool代码之前我会问这里.

我正在使用Astyanax客户端从Cassandra数据库中读取数据.

我在Cassandra数据库中有大约一百万个唯一行.我有一个包含四个节点的交叉主机托管集群.

这些是我的四个节点:

  node1:9160
  node2:9160
  node3:9160
  node4:9160

我启用了KeyCaching,并且还启用了SizeTieredCompaction策略.

我有一个多线程的客户端程序,它将使用Astyanax客户端从Cassandra数据库读取数据,并且我运行20个线程.如果我使用20个线程运行我的客户端程序,那么从Cassandra数据库读取数据的性能会降低.

因此,我想到的第一件事就是可能存在与Cassandra连接的争用(他们是否使用了池,如果有的话,维护了多少连接)？我使用以下代码使用Astyanax客户端建立连接.

private CassandraAstyanaxConnection() {
    context = new AstyanaxContext.Builder()
    .forCluster(ModelConstants.CLUSTER)
    .forKeyspace(ModelConstants.KEYSPACE)
    .withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
        .setDiscoveryType(NodeDiscoveryType.RING_DESCRIBE)
    )
    .withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("MyConnectionPool")
        .setPort(9160)
        .setMaxConnsPerHost(1)
        .setSeeds("nod1:9160,node2:9160,node3:9160,node4:9160")
    )
    .withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
        .setCqlVersion("3.0.0")
        .setTargetCassandraVersion("1.2"))
    .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
    .buildKeyspace(ThriftFamilyFactory.getInstance());

    context.start();
    keyspace = context.getEntity();

    emp_cf = ColumnFamily.newColumnFamily(
        ModelConstants.COLUMN_FAMILY,
        StringSerializer.get(),
        StringSerializer.get());
}

我是否需要在上面的代码中进行任何类型的更改以提高性能？

这种方法有什么作用？

   setMaxConnsPerHost(1)

我是否需要增加它以提高性能？我有四个节点,所以我应该将其更改为4？

并且setMaxConns(20)方法会调用吗？我是否还需要添加它以提高性能？因为我将用多个线程运行我的程序.

我TOKEN_AWARE在com.netflix.astyanax.connectionpool.NodeDiscoveryType中为Cassandra 找到了Astyanax客户端的枚举值,并且我试图了解它的作用？

package com.netflix.astyanax.connectionpool;

public enum NodeDiscoveryType {
    /**
     * Discover nodes exclusively from doing a ring describe
     */
    RING_DESCRIBE,

    /**
     * Discover nodes exclusively from an external node discovery service
     */
    DISCOVERY_SERVICE,

    /**
     * Intersect ring describe and nodes from an external service. This solve
     * the multi-region ring describe problem where ring describe returns nodes
     * from other regions.
     */
    TOKEN_AWARE,

    /**
     * Use only nodes in the list of seeds
     */
    NONE
} …

我们的 CF 仅有大约 1000 行。有没有办法用 astyanax 读取所有 1000 行？节俭甚至支持这一点吗？

谢谢，迪恩

我使用的是astyanax连接池,定义如下:

ipSeeds = "LOAD_BALANCER_HOST:9160";
conPool.setSeeds(ipSeeds)
.setDiscoveryType(NodeDiscoveryType.TOKEN_AWARE)
.setConnectionPoolType(ConnectionPoolType.TOKEN_AWARE);

但是,我的群集有4个节点,我有8个客户端机器连接在它上面.LOAD_BALANCER_HOST将请求转发给我的四个节点之一.

在客户端节点上,我有:

$netstat -an | grep 9160 | awk '{print $5}' | sort |uniq -c
    235 node1:9160
    680 node2:9160
      4 node3:9160
      4 node4:9160

因此,虽然ConnectionPoolType是TOKEN_AWARE,我的客户端似乎主要连接到node2,有时连接到node1,但几乎从不连接到节点3和4.
问题是:为什么会发生这种情况？难道令牌识别连接池不应该查询节点列表的环并使用循环算法连接到所有活动节点吗？

我正在尝试使用以下astyanax代码从cassandra中读取Set类型的列.

val genres = col.getColumnByName("genres")
val genValue = genres.getValue(new SetSerializer[String](UTF8Type.instance))

我在Astyanax文档中也找到了类似的代码

https://github.com/Netflix/astyanax/wiki/Collections

但我得到错误

org.apache.cassandra.serializers.MarshalException: Unexpected extraneous bytes after set value
    at org.apache.cassandra.serializers.SetSerializer.deserialize(SetSerializer.java:64)
    at org.apache.cassandra.serializers.SetSerializer.deserialize(SetSerializer.java:27)
    at org.apache.cassandra.db.marshal.AbstractType.compose(AbstractType.java:142)
    at com.netflix.astyanax.serializers.SetSerializer.fromByteBuffer(SetSerializer.java:32)

我的表定义是

CREATE TABLE movielens_small.movies (
    id uuid PRIMARY KEY,
    avg_rating float,
    genres set<text>,
    name text,
    release_date date,
    url text,
    video_release_date date
) WITH bloom_filter_fp_chance = 0.01

我可以轻松地在cqlsh中选择一个查询.所以我认为db没有问题.

编辑::我也试过了

val myset = ListType.getInstance(UTF8Type.instance)
val genValue = myset.compose(genres.getByteBufferValue) 

但它会抛出相同的错误,即存在意外的无关字节.

Edit2 ::我也试过了

val genValue = new String(genres.getByteBufferValue.array(), "UTF-8")

这不会引发错误,我可以看到数据......但它就像乱码.

Edit3 ::我也试过了

val setSer = new …

我知道我们可以用这样的东西向Cassandra插入单个记录(下面的例子取自这里):

final String INSERT_STATEMENT = "INSERT INTO employees (empID, deptID, first_name, last_name) VALUES (?, ?, ?, ?);";

result = keyspace
    .prepareQuery(CQL3_CF)
    .withCql(INSERT_STATEMENT)
    .asPreparedStatement()
    .withIntegerValue(222)
    .withIntegerValue(333)
    .withStringValue("Eric")
    .withStringValue("Cartman")
    .execute();

是否可以使用Astyanax的cql3 API(如JDBC的executeBatch)进行批量插入(用于多个记录)？

注意:使用Astyanax的MutationBatch(基于Thrift,而不是CQL)对我来说似乎不是一个选择,因为我遇到了与此问题相同的问题.

所以这就是我的困境:我有一个域模型,在scala中有一堆case类,比如User和Organization.在我的数据访问层(dao,存储库等)中,我使用的是astyanax(来自netflix的java库),它的实体持久化将对象保存到cassandra列族.

这是我的cassandra/astyanax支持的DAO的一些示例代码(是的,我知道我需要做更多的scala-ish,但我还在学习=))

通过这个啰嗦的描述看完后,我基本上想看看,为什么在参数列表中注释的瓦尔斯将无法正常工作时,Java并getDeclaredAnnotations()在Field我会讨厌要回去和重构一切,所以我可以用它的持留保存实体(即manager.put(entity))非常简单.如果我想继续使用case类,以便我可以使用更多不可变的样式scala和Lensscalaz,那么我将不得不更新DAO并手动执行所有持久化操作,这可以真正消磨时间.

所以,如果有人知道我没有看到的东西,请告诉我!提前感谢您花时间阅读本文.

场景1 - 案例类

Astyanax没有拿起注释@Id val

@Entity
case class Organization(@Id @Column(name = "id") override val id: Option[UUID] = None,
                        @Column(name = "created_on") override val createdOn: Option[Date] = None,
                        @Column(name = "modified_on") override val modifiedOn: Option[Date] = None,
                        @Column(name = "name") name: Option[String] = None,
                        @Column(name = "is_paid_account") isPaidAccount: Boolean = false) extends IdBaseEntity[UUID](id, createdOn, modifiedOn)

场景2 - 具有伴随对象的类或没有伴随对象的类

Astyanax没有拿起@Id注释 val

@Entity
class Organization(@Id @Column(name = …

• 我有一个3节点C *集群。
• C *客户端的读取一致性级别设置为QUORUM。
• 当集群中的一个节点关闭时，我收到UnavailableException来响应读取的查询

为什么？由3个节点组成的群集的法定人数为2，因此它应处理一个节点的中断。

更多细节：

Cassandra版本：

ReleaseVersion: 1.1.6

键空间和列族的配置：

Keyspace: QuestionAnswerService:
  Replication Strategy: org.apache.cassandra.locator.NetworkTopologyStrategy
  Durable Writes: true
    Options: [datacenter1:2]
  Column Families:
  //...
  ColumnFamily: answersByQuestion
      Key Validation Class: org.apache.cassandra.db.marshal.BytesType
      Default column value validator: org.apache.cassandra.db.marshal.BytesType
      Columns sorted by: org.apache.cassandra.db.marshal.BytesType
      GC grace seconds: 864000
      Compaction min/max thresholds: 4/32
      Read repair chance: 1.0
      DC Local Read repair chance: 0.0
      Populate IO Cache on flush: false
      Replicate on write: true
      Caching: KEYS_ONLY
      Bloom Filter FP chance: default
      Compaction Strategy: org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy
      Compression Options: …

当您使用cqlshCassandra时,您可以进行describe查询以获取表的信息,例如:

DESCRIBE TABLE emp;

它会给你一些类似的东西:

CREATE TABLE emp (
  empid int PRIMARY KEY,
  deptid int,
  description text
) ...
....

那么如何使用Astyanax对CQL的支持来查询它.我能够使用以下方法查询简单SELECT语句:

OperationResult<CqlResult<String, String>> result
            = keyspace.prepareQuery(empColumnFamily)
              .withCql("Select * from emp;")
              .execute();

但这不适用于DESCRIBE陈述.

PD:我真的这样做是为了得到DATA TYPES表格,稍后解析它并获得例如int, int, text,所以如果你有不同的方法来获得它们,那么它可能很棒.

此查询select column, validator from system.schema_columns;不起作用,因为它不返回组合键.

我正在尝试使用Astyanax驱动程序列出Cassandra中的列族.它列出了键空间OK,但输出中缺少许多列族.

我有一个简单的程序:

import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.Cluster;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.ddl.ColumnFamilyDefinition;
import com.netflix.astyanax.ddl.KeyspaceDefinition;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;

public class App {

  public static void main(String[] args) throws Exception {

    ConnectionPoolConfigurationImpl cpool = new ConnectionPoolConfigurationImpl("ConnectionPool")
        .setPort(9160)
        .setSeeds("localhost");

    AstyanaxConfigurationImpl astyanaxConfiguration = new AstyanaxConfigurationImpl();
    AstyanaxContext.Builder ctxBuilder = new AstyanaxContext.Builder();
    ctxBuilder.forCluster("Cluster")
        .withAstyanaxConfiguration(astyanaxConfiguration)
        .withConnectionPoolConfiguration(cpool)
        .withConnectionPoolMonitor(new CountingConnectionPoolMonitor());

    AstyanaxContext<Cluster> clusterContext = ctxBuilder.buildCluster(ThriftFamilyFactory.getInstance());
    clusterContext.start();
    Cluster cluster = clusterContext.getClient();

    for (KeyspaceDefinition ksDef : cluster.describeKeyspaces()) {
      List<ColumnFamilyDefinition> cfDefList = ksDef.getColumnFamilyList();
      System.out.println("there are " + cfDefList.size() + " column families …