NoSql 之 Cassandra 查询

java操作Cassandra之前得操作一下步骤:
1.启动cassandra.
2.进入到${cassandra-home}\bin下,执行

cassandra-cli -host localhost -port 9160

3.创建空间:

create keyspace wingware;

4.切换到创建好的空间中:

use wingware;

5.创建column family:
create column family wing;

运行以下代码需要${cassandra-home}\lib下的*.jar添加到当前的classpath下.
SampleOne.java

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

public class SampleOne {
	static Cassandra.Client cassandraClient;
	static TTransport socket;

	private static void init(String keySpace) throws InvalidRequestException, TException {
		String server = "localhost";
		int port = 9160;
		/* 首先指定cassandra server的地址 */
		socket = new TSocket(server, port);
		System.out.println(" connected to " + server + ":" + port + ".");
		TFramedTransport transport = new TFramedTransport(socket);
		/* 指定通信协议为二进制流协议 */
		TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport);
		cassandraClient = new Cassandra.Client(binaryProtocol);
		/* 建立通信连接 */
		socket.open();
		cassandraClient.set_keyspace(keySpace);
	}

	public static void main(String[] args) throws TException, TimedOutException, InvalidRequestException, UnavailableException, NotFoundException {
		/* 选择需要操作的Keyspaces， 可以理解成数据库的表 */
		String keyspace = "wingware";

		/* 初始化连接 */
		init(keyspace);

		/* 创建一个Table Name */
		String columnFamily = "wing";
		
		String tablename = "tablename";

		/* 插入一条记录 */
		insertOrUpdate(columnFamily, tablename, "xiexie2", "dddddddd", System.currentTimeMillis());
		/* 删除一条记录 */
		// delete(keyspace,tableName,row,"name",System.currentTimeMillis());
		/* 获取一条记录 (由于插入和删除是同一条记录,有可能会检索不到哦!请大家主意! */
		Column column = getByColumn(columnFamily, tablename, "key", System.currentTimeMillis());

		System.out.println("read column " + columnFamily);
		System.out.println("column name " + ":" + new String(column.getName()));
		System.out.println("column value" + ":" + new String(column.getValue()));
		System.out.println("column timestamp" + ":" + (column.timestamp));
		column = getByColumn(columnFamily, tablename, "xiexie", System.currentTimeMillis());

		System.out.println("read row " + columnFamily);
		System.out.println("column name " + ":" + new String(column.getName()));
		System.out.println("column value" + ":" + new String(column.getValue()));
		System.out.println("column timestamp" + ":" + (column.timestamp));
		close();
	}

	/**
	 * 插入记录
	 */
	public static void insertOrUpdate(String columnFamily, String tableName, String ColumnName, String ColumnValue, long timeStamp) throws TException, TimedOutException, InvalidRequestException,
			UnavailableException, NotFoundException {
		/* 数据所在的行标 */

		/* 创建一个column path */
		ColumnParent parent = new ColumnParent(columnFamily);
		Column col = new Column();
		col.setName(ColumnName.getBytes());
		col.setValue(ColumnValue.getBytes());

		/*
		 * 执行插入操作，指定keysapce, row, col, 和数据内容， 后面两个参数一个是timestamp，
		 * 另外一个是consistency_level timestamp是用来做数据一致性保证的，
		 * 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo
		 */
		cassandraClient.insert(Translater.toByteBuffer(tableName), parent, col, ConsistencyLevel.ONE);
	}

	/**
	 * 删除记录
	 */
	public static void delete(String columnFamily, String tablename, String ColumnName, long timeStamp) throws TException, TimedOutException, InvalidRequestException, UnavailableException,
			NotFoundException {
		/* 选择需要操作的Keyspaces， 存放数据表所在的空间位置 */
		/* 数据所在的行标 */
		/* 创建一个column path */
		ColumnPath col = new ColumnPath(columnFamily);
		col.setColumn(ColumnName.getBytes());

		/*
		 * 执行删除操作，指定keysapce, row, col， 后面两个参数一个是timestamp，
		 * 另外一个是consistency_level timestamp是用来做数据一致性保证的，
		 * 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo
		 */
		cassandraClient.remove(Translater.toByteBuffer(tablename), col, System.currentTimeMillis(), ConsistencyLevel.ONE);
	}

	/**
	 * 获取数据
	 */
	public static Column getByColumn(String columnFamily, String tablename, String ColumnName, long timeStamp) throws TException, TimedOutException, InvalidRequestException, UnavailableException,
			NotFoundException {

		/* 创建一个columnFamily */
		ColumnPath col = new ColumnPath(columnFamily);
		col.setColumn(ColumnName.getBytes());
		/*
		 * 执行查询操作，指定keysapce, row, col， timestamp timestamp是用来做数据一致性保证的，
		 * 而consistency_level是用来控制数据分布的策略，前者的理论依据是bigtable, 后者的理论依据是dynamo
		 */
		ColumnOrSuperColumn superColumn = cassandraClient.get(Translater.toByteBuffer(tablename), col, ConsistencyLevel.ONE);
		System.out.println(">>>>>>>>>>>>>>>>"+superColumn);
		
		Column column = cassandraClient.get(Translater.toByteBuffer(tablename), col, ConsistencyLevel.ONE).column;
		return column;
	}

	/**
	 * 关闭当前的远程访问连接
	 */
	public static void close() {
		socket.close();
	}
}

Translater.java

import java.nio.ByteBuffer;  
import java.nio.CharBuffer;  
import java.nio.charset.Charset;  
import java.nio.charset.CharsetDecoder;  
import java.nio.charset.CharsetEncoder;  
  
public class Translater {  
  
    public static Charset charset = Charset.forName("UTF-8");  
    public static CharsetEncoder encoder = charset.newEncoder();  
    public static CharsetDecoder decoder = charset.newDecoder();  
  
    public static ByteBuffer toByteBuffer(String msg) {  
        try {  
            return encoder.encode(CharBuffer.wrap(msg));  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
        return null;  
    }  
  
    public static String toString(ByteBuffer buffer) {  
        String data = "";  
        try {  
            int old_position = buffer.position();  
            data = decoder.decode(buffer).toString();  
            // reset buffer's position to its original so it is not altered:  
            buffer.position(old_position);  
        } catch (Exception e) {  
            e.printStackTrace();  
            return "";  
        }  
        return data;  
    }  
} 

keyspace+columnFamily+tablename+column;
若column不存在,会抛出,而不是返回null.
Exception in thread "main" NotFoundException()
at org.apache.cassandra.thrift.Cassandra$get_result.read(Cassandra.java:5932)
at org.apache.cassandra.thrift.Cassandra$Client.recv_get(Cassandra.java:489)
at org.apache.cassandra.thrift.Cassandra$Client.get(Cassandra.java:462)



cassandra的命令:

List of all CLI commands:
?                                                          Display this message.
help;                                                         Display this help.
help <command>;                         Display detailed, command-specific help.
connect <hostname>/<port> (<username> '<password>')?; Connect to thrift service.
use <keyspace> [<username> 'password'];                    Switch to a keyspace.
describe keyspace (<keyspacename>)?;                          Describe keyspace.
exit;                                                                  Exit CLI.
quit;                                                                  Exit CLI.
describe cluster;                             Display information about cluster.
show cluster name;                                         Display cluster name.
show keyspaces;                                          Show list of keyspaces.
show api version;                                       Show server API version.
create keyspace <keyspace> [with <att1>=<value1> [and <att2>=<value2> ...]];
                Add a new keyspace with the specified attribute(s) and value(s).
update keyspace <keyspace> [with <att1>=<value1> [and <att2>=<value2> ...]];
                 Update a keyspace with the specified attribute(s) and value(s).
create column family <cf> [with <att1>=<value1> [and <att2>=<value2> ...]];
        Create a new column family with the specified attribute(s) and value(s).
update column family <cf> [with <att1>=<value1> [and <att2>=<value2> ...]];
            Update a column family with the specified attribute(s) and value(s).
drop keyspace <keyspace>;                                     Delete a keyspace.
drop column family <cf>;                                 Delete a column family.
get <cf>['<key>'];                                       Get a slice of columns.
get <cf>['<key>']['<super>'];                        Get a slice of sub columns.
get <cf> where <column> = <value> [and <column> > <value> and ...] [limit int];
get <cf>['<key>']['<col>'] (as <type>)*;                     Get a column value.
get <cf>['<key>']['<super>']['<col>'] (as <type>)*;      Get a sub column value.
set <cf>['<key>']['<col>'] = <value> (with ttl = <secs>)*;         Set a column.
set <cf>['<key>']['<super>']['<col>'] = <value> (with ttl = <secs>)*;
                                                               Set a sub column.
del <cf>['<key>'];                                                Delete record.
del <cf>['<key>']['<col>'];                                       Delete column.
del <cf>['<key>']['<super>']['<col>'];                        Delete sub column.
count <cf>['<key>'];                                    Count columns in record.
count <cf>['<key>']['<super>'];                 Count columns in a super column.
truncate <column_family>;                      Truncate specified column family.
assume <column_family> <attribute> as <type>;
              Assume a given column family attributes to match a specified type.
consistencylevel as <level>;
                  Change the consistency level for set,get, and list operations.
list <cf>;                                   List all rows in the column family.
list <cf>[<startKey>:];
                       List rows in the column family beginning with <startKey>.
list <cf>[<startKey>:<endKey>];
        List rows in the column family in the range from <startKey> to <endKey>.
list ... limit N;                                   Limit the list results to N.

评论

1 楼 zendly 2015-09-12  

mark一下