cassandra - 如何读取可能/可能不存在的复合列，获取其值以更新它并将其保存回来？

Question

我有一个复合列(Int32Type,BytesType,AsciiType)，我需要读取它的值(基于条件)、修改它并将它保存回来(某物像手动计数器列)。

我正在查询的复合列可能存在也可能不存在。

在 Hector 中最好的方法是什么？

Answer 1

我不能保证以下解决方案是最好的，但它具有创建复合列等基本功能。它基本上进行读写，本质上是内联到“我需要读取它的值(基于标准)，修改它并将它保存回来(类似于手动计数器列)。” .但是我觉得有了这个样例代码，应该可以作为一个基础，在这里和那里进行改进，从而成为最好的。 :-) 有空闲时间会彻底测试它。话虽如此，以下是我的建议。

package com.hector.dataTypes;

import java.util.Iterator;

import me.prettyprint.cassandra.serializers.ByteBufferSerializer;
import me.prettyprint.cassandra.serializers.CompositeSerializer;
import me.prettyprint.cassandra.serializers.IntegerSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.cassandra.service.ColumnSliceIterator;
import me.prettyprint.cassandra.service.ThriftCluster;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.AbstractComposite.ComponentEquality;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.Composite;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnType;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.exceptions.HectorException;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SliceQuery;

import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.utils.ByteBufferUtil;

import com.google.common.base.Joiner;

/**
 *
 * @author jasonw
 *
 */
public class CompositeExample
{
    private String m_node;
    private String m_keyspace;
    private String m_column_family;
    private ThriftCluster m_cassandraCluster;
    private CassandraHostConfigurator m_cassandraHostConfigurator;
    private Mutator<String> mutator;
    private SliceQuery<String, Composite, String> sliceQuery;

    public CompositeExample(String p_node, String p_keyspace, String p_column_family, String p_cluster)
    {
        m_node = p_node;
        m_keyspace = p_keyspace;
        m_column_family = p_column_family;
        m_cassandraHostConfigurator = new CassandraHostConfigurator(m_node);
        m_cassandraCluster = new ThriftCluster(p_cluster, m_cassandraHostConfigurator);
        Cluster cluster = HFactory.getOrCreateCluster(p_cluster, m_cassandraHostConfigurator);
        Keyspace keyspace = HFactory.createKeyspace(m_keyspace, cluster);
        mutator = HFactory.createMutator(keyspace, StringSerializer.get());
        sliceQuery = HFactory.createSliceQuery(keyspace, StringSerializer.get(), CompositeSerializer.get(), StringSerializer.get());
    }

    public boolean createCompositeColumn(String... p_new_columns)
    {
        try
        {
            ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition(m_keyspace, m_column_family, ComparatorType.COMPOSITETYPE);
            cfDef.setColumnType(ColumnType.STANDARD);
            cfDef.setComparatorTypeAlias("(".concat(Joiner.on(",").join(p_new_columns)).concat(")"));
            cfDef.setKeyValidationClass("UTF8Type");
            cfDef.setDefaultValidationClass("UTF8Type");
            m_cassandraCluster.addColumnFamily(cfDef, true);           
            return true;
        }
        catch (HectorException e)
        {
            e.printStackTrace();
        }

        return false;

    }

    public boolean saveColumn(String p_field_one, String p_field_two, String p_field_three)
    {

        try
        {
            Composite c = new Composite();
            c.addComponent(Int32Type.instance.fromString(p_field_one), ByteBufferSerializer.get());
            c.addComponent(ByteBufferUtil.bytes(p_field_two), ByteBufferSerializer.get());
            c.addComponent(p_field_three, StringSerializer.get());
            HColumn<Composite, String> col = HFactory.createColumn(c, "composite_value", CompositeSerializer.get(), StringSerializer.get());
            mutator.addInsertion("key", m_column_family, col);
            mutator.execute();                   
            return true;

        }
        catch (HectorException e)
        {
            e.printStackTrace();           
        }

        return false;

    }

    public boolean readColumn(String p_key, int p_column_number, ComponentEquality p_equality, int p_value)
    {
        if (p_column_number < 0 || p_column_number > 2)
        {
            return false;
        }

        try
        {
            sliceQuery.setColumnFamily(m_column_family);
            sliceQuery.setKey(p_key);       


            Composite start = new Composite();
            start.addComponent(0, p_value, p_equality);
            Composite end = new Composite();
            end.addComponent(0, p_value, ComponentEquality.GREATER_THAN_EQUAL);         
            sliceQuery.setRange(start, end, false, 1000);

            QueryResult<ColumnSlice<Composite, String>> qr = sliceQuery.execute();

            System.out.println("size = " + qr.get().getColumns().size());

            Iterator<HColumn<Composite, String>> iter = qr.get().getColumns().iterator();

            while (iter.hasNext())
            {
                HColumn<Composite, String> column = iter.next();
                System.out.print(column.getName().get(0, IntegerSerializer.get()));
                System.out.print(":");
                System.out.print(column.getName().get(1, StringSerializer.get()));
                System.out.print(":");
                System.out.print(column.getName().get(2, StringSerializer.get()));
                System.out.println("=" + column.getValue());
            }

            return true;
        }
        catch (HectorException e)
        {
            e.printStackTrace();
        }
        catch (Exception why)
        {
            why.printStackTrace();
        }

        return false;

    }

    public static void main(String[] args)
    {
        boolean isSuccess = false;
        String node_ip = "192.168.0.1";
        String keyspace_name = "mykeyspace";
        String column_family_name = "compositecf";
        String cluster_name = "Test Cluster";

        CompositeExample test1 = new CompositeExample(node_ip, keyspace_name, column_family_name, cluster_name);

        isSuccess = test1.createCompositeColumn("Int32Type", "BytesType", "AsciiType");       
        if (!isSuccess)
        {
            System.err.println("failed to create cf");
            System.exit(-1);
        }

        isSuccess = test1.saveColumn("1027", "blablabla", "this is ascii field");
        if (!isSuccess)
        {
            System.err.println("failed to write");
            System.exit(-1);
        }

        isSuccess = test1.readColumn("key", 0, ComponentEquality.EQUAL, 1027);
        if (!isSuccess)
        {
            System.err.println("failed to read");
            System.exit(-1);
        }


    }

}