import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
public List<String> QueryOneMinute(String min) {
	List<String> userList = new ArrayList<String>();
	try {
		List<ColumnOrSuperColumn> results = client.get_slice(Utils
				.toByteBuffer(min), columnParent, predicate,
				ConsistencyLevel.ONE);
		for (ColumnOrSuperColumn cc : results) {
			SuperColumn superColumn = cc.getSuper_column();
			List<Column> list = superColumn.getColumns();
			for (Column c : list) {
				String columnName = new String(c.getName(), "UTF-8");
				if (columnName.equals("username")) {
					String value = new String(c.getValue(), "UTF-8");
					if (!userList.contains(value)) {
						userList.add(value);
					}
				}
			}
		}
	} catch (Exception e) {
		System.out.println(e);
	}
	return userList;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
public List<String> QueryOneMinute(String min){
	List<String> bookList = new ArrayList<String>();
	try {
		List<ColumnOrSuperColumn> results = client.get_slice(Utils
				.toByteBuffer(min), columnParent, predicate,
				ConsistencyLevel.ONE);
		for (ColumnOrSuperColumn cc : results) {
			SuperColumn superColumn = cc.getSuper_column();
			List<Column> list = superColumn.getColumns();
			for (Column c : list) {
				String columnName = new String(c.getName(), "UTF-8");
				if (columnName.equals("bookno")) {
					String value = new String(c.getValue(), "UTF-8");
					if (!bookList.contains(value)) {
						bookList.add(value);
					}
				}
			}
		}
	} catch (Exception e) {
		System.out.println(e);
	}
	return bookList;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private void insertPOISpringTraining() throws Exception {
	Map<ByteBuffer, Map<String, List<Mutation>>> outerMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
	List<Mutation> columnsToAdd = new ArrayList<Mutation>();

	long timestamp = System.nanoTime();
	String keyName = "Spring Training";
	Column descCol = new Column(bytes("desc"));
	Column phoneCol = new Column(bytes("phone"));

	List<Column> cols = new ArrayList<Column>();
	cols.add(descCol);
	cols.add(phoneCol);

	Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

	Mutation columns = new Mutation();
	ColumnOrSuperColumn descCosc = new ColumnOrSuperColumn();
	SuperColumn sc = new SuperColumn();
	sc.name = bytes(CAMBRIA_NAME);
	sc.columns = cols;

	descCosc.super_column = sc;
	columns.setColumn_or_supercolumn(descCosc);

	columnsToAdd.add(columns);

	String superCFName = "PointOfInterest";
	ColumnPath cp = new ColumnPath();
	cp.column_family = superCFName;
	cp.setSuper_column(CAMBRIA_NAME.getBytes());
	cp.setSuper_columnIsSet(true);

	innerMap.put(superCFName, columnsToAdd);
	outerMap.put(bytes(keyName), innerMap);

	client.batch_mutate(outerMap, CL);

	LOG.debug("Done inserting Spring Training.");
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private void insertPOICentralPark() throws Exception {

		Map<ByteBuffer, Map<String, List<Mutation>>> outerMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
		List<Mutation> columnsToAdd = new ArrayList<Mutation>();

		long ts = System.nanoTime();
		String keyName = "Central Park";
		Column descCol = new Column(bytes("desc"));

		// no phone column for park

		List<Column> cols = new ArrayList<Column>();
		cols.add(descCol);

		Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

		Mutation columns = new Mutation();
		ColumnOrSuperColumn descCosc = new ColumnOrSuperColumn();
		SuperColumn waldorfSC = new SuperColumn();
		waldorfSC.name = bytes(WALDORF_NAME);
		waldorfSC.columns = cols;

		descCosc.super_column = waldorfSC;
		columns.setColumn_or_supercolumn(descCosc);

		columnsToAdd.add(columns);

		String superCFName = "PointOfInterest";
		ColumnPath cp = new ColumnPath();
		cp.column_family = superCFName;
		cp.setSuper_column(WALDORF_NAME.getBytes());
		cp.setSuper_columnIsSet(true);

		innerMap.put(superCFName, columnsToAdd);
		outerMap.put(bytes(keyName), innerMap);

		client.batch_mutate(outerMap, CL);

		LOG.debug("Done inserting Central Park.");
	}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private IColumn unthriftifySuper(SuperColumn super_column) {
  org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(
      super_column.name, subComparator);
  for (Column column : super_column.columns) {
    sc.addColumn(unthriftifySimple(column));
  }
  return sc;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private IColumn unthriftifySuperCounter(CounterSuperColumn superColumn) {
  org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(
      superColumn.name, subComparator);
  for (CounterColumn column : superColumn.columns) {
    sc.addColumn(unthriftifyCounter(column));
  }
  return sc;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private IColumn unthriftifySuper(SuperColumn super_column)
{
    org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(super_column.name, subComparator);
    for (Column column : super_column.columns)
    {
        sc.addColumn(unthriftifySimple(column));
    }
    return sc;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private IColumn unthriftifySuperCounter(CounterSuperColumn superColumn)
{
    org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(superColumn.name, subComparator);
    for (CounterColumn column : superColumn.columns)
        sc.addColumn(unthriftifyCounter(column));
    return sc;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
/**
 * Query and Check item is exist or not
 * 
 * return times of item
 */
public int QueryIpUser(String ip, String name){
	String IP_KEY = ip;
	String NAME_SUPER_KEY = name;
	String COLUMN_NAME = "times";
	String COLUMN_FAMILY = "IpUser";
	
	try{
		ColumnPath columnPath = new ColumnPath();
		columnPath.column_family = COLUMN_FAMILY;
		columnPath.super_column = this.cassandraUtil.toByteBuffer(NAME_SUPER_KEY);
		
		ColumnOrSuperColumn columnOrSuperColumn = client.get(this.cassandraUtil.toByteBuffer(IP_KEY), columnPath, ConsistencyLevel.ONE);

		SuperColumn superColumn = columnOrSuperColumn.getSuper_column();
		
		List<Column> columns = superColumn.getColumns();
		
		for(Column col : columns) {
			String title = new String(col.getName(), "UTF-8") ;
			
			if(title.equals(COLUMN_NAME)){
				return Integer.parseInt(new String(col.getValue(), "UTF-8"));     // get "times" column and return 
			}
		}
	}catch(Exception e){
		return 0;
	}

	return 0;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
/**
 * Query and Check item is exist or not
 * 
 * return times of item
 */
public int QueryUserIp(String ip, String name){
	String NAME_KEY = name;
	String IP_SUPER_KEY = ip;
	String COLUMN_NAME = "times";
	String COLUMN_FAMILY = "UserIp";
	
	try{
		ColumnPath columnPath = new ColumnPath();
		columnPath.column_family = COLUMN_FAMILY;
		columnPath.super_column = this.cassandraUtil.toByteBuffer(IP_SUPER_KEY);
		
		ColumnOrSuperColumn columnOrSuperColumn = client.get(this.cassandraUtil.toByteBuffer(NAME_KEY), columnPath, ConsistencyLevel.ONE);

		SuperColumn superColumn = columnOrSuperColumn.getSuper_column();
		
		List<Column> columns = superColumn.getColumns();
		
		for(Column col : columns) {
			String title = new String(col.getName(), "UTF-8") ;
			if(title.equals(COLUMN_NAME)){
				return Integer.parseInt(new String(col.getValue(), "UTF-8"));     // get "times" column and return 
			}
		}
	}catch(Exception e){
		return 0;
	}
	
	return 0;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
/**
 * Query and Check BookIp
 * 
 * return times of column is exist, 0 or not
 */
public int QueryBookIp(String bookid, String ip) {
	String BOOKID_KEY = bookid;
	String IP_SUPER_KEY = ip;
	String COLUMN_NAME = "times";
	String COLUMN_FAMILY = "BookIp";

	try {
		ColumnPath columnPath = new ColumnPath();
		columnPath.column_family = COLUMN_FAMILY;
		columnPath.super_column = this.cassandraUtil.toByteBuffer(IP_SUPER_KEY);

		ColumnOrSuperColumn columnOrSuperColumn = client.get(this.cassandraUtil.toByteBuffer(BOOKID_KEY), columnPath,ConsistencyLevel.ONE);

		SuperColumn superColumn = columnOrSuperColumn.getSuper_column();

		List<Column> columns = superColumn.getColumns();


		for (Column col : columns) {
			String title = new String(col.getName(), "UTF-8");
			if (title.equals(COLUMN_NAME)) {
				return Integer.parseInt(new String(col.getValue(), "UTF-8"));
			}
		}

	} catch (Exception e) {
		return 0;
	}

	return 0;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
/**
 * Query and Check BookUser
 * 
 * return times of column is exist, 0 or not
 */
public int QueryBookUser(String bookid, String user) {
	String BOOKID_KEY = bookid;
	String USER_SUPER_KEY = user;
	String COLUMN_NAME = "times";
	String COLUMN_FAMILY = "BookUser";
	
	try{
		ColumnPath columnPath = new ColumnPath();
		columnPath.column_family = COLUMN_FAMILY;
		columnPath.super_column = this.cassandraUtil.toByteBuffer(USER_SUPER_KEY);
		
		ColumnOrSuperColumn columnOrSuperColumn = client.get(this.cassandraUtil.toByteBuffer(BOOKID_KEY), columnPath, ConsistencyLevel.ONE);
		
		SuperColumn superColumn = columnOrSuperColumn.getSuper_column();
		
		List<Column> columns = superColumn.getColumns();
		
		for(Column col : columns) {
			String title = new String(col.getName(), "UTF-8");
			if(title.equals(COLUMN_NAME)) {
				return Integer.parseInt(new String(col.getValue(), "UTF-8"));
			}
		}
		
	}catch(Exception e) {
		return 0;
	}
	
	return 0;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
/**
 * Query and Check IpBookPage
 * 
 * return times of column is exist, 0 or not
 */
public int QueryIpBookPage(String ip, String book) {
	String IP_KEY = ip;
	String BOOKID_SUPER_KEY = book;
	String COLUMN_FAMILY = "IpBookPage";
	String COLUMN_NAME = "times";
	
	try{
		ColumnPath columnPath = new ColumnPath();
		columnPath.column_family = COLUMN_FAMILY;
		columnPath.super_column = this.cassandraUtil.toByteBuffer(BOOKID_SUPER_KEY);
		
		ColumnOrSuperColumn columnOrSuperColumn = client.get(this.cassandraUtil.toByteBuffer(IP_KEY), columnPath, ConsistencyLevel.ONE);
		
		SuperColumn superColumn = columnOrSuperColumn.getSuper_column();
		
		List<Column> columns = superColumn.getColumns();
		
		for(Column col : columns) {
			String title = new String(col.getName(), "UTF-8");
			if(title.equals(COLUMN_NAME)) {
				return Integer.parseInt(new String(col.getValue(), "UTF-8"));
			}
		}
		
	}catch(Exception e){
		return 0;
	}
	
	return 0;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
/**
 * Query and Check UserBookPage
 * 
 * return times of column is exist, 0 or not
 */
public int QueryUserBookPage(String user, String book) {
	String USER_KEY = user;
	String BOOK_SUPER_KEY = book;
	String COLUMN_FAMILY = "UserBookPage";
	String COLUMN_NAME = "times";
	
	try{
		ColumnPath columnPath = new ColumnPath();
		columnPath.column_family = COLUMN_FAMILY;
		columnPath.super_column = this.cassandraUtil.toByteBuffer(BOOK_SUPER_KEY);
		
		ColumnOrSuperColumn columnOrSuperColumn = client.get(this.cassandraUtil.toByteBuffer(USER_KEY), columnPath, ConsistencyLevel.ONE);
		
		SuperColumn superColumn = columnOrSuperColumn.getSuper_column();
		
		List<Column> columns = superColumn.getColumns();
		
		for(Column col : columns) {
			String title = new String(col.getName(), "UTF-8");
			if(title.equals(COLUMN_NAME)) {
				return Integer.parseInt(new String(col.getValue(), "UTF-8"));
			}
		}
		
	}catch(Exception e) {
		return 0;
	}
	
	return 0;
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private void insertPOIPhoenixZoo() throws Exception {

		Map<ByteBuffer, Map<String, List<Mutation>>> outerMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
		List<Mutation> columnsToAdd = new ArrayList<Mutation>();

		long ts = System.currentTimeMillis();
		String keyName = "Phoenix Zoo";
		Column descCol = new Column(bytes("desc"));

		Column phoneCol = new Column(bytes("phone"));

		List<Column> cols = new ArrayList<Column>();
		cols.add(descCol);
		cols.add(phoneCol);

		Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

		String cambriaName = "Cambria Suites Hayden";

		Mutation columns = new Mutation();
		ColumnOrSuperColumn descCosc = new ColumnOrSuperColumn();
		SuperColumn sc = new SuperColumn();
		sc.name = bytes(cambriaName);
		sc.columns = cols;

		descCosc.super_column = sc;
		columns.setColumn_or_supercolumn(descCosc);

		columnsToAdd.add(columns);

		String superCFName = "PointOfInterest";
		ColumnPath cp = new ColumnPath();
		cp.column_family = superCFName;
		cp.setSuper_column(cambriaName.getBytes());
		cp.setSuper_columnIsSet(true);

		innerMap.put(superCFName, columnsToAdd);
		outerMap.put(bytes(keyName), innerMap);

		client.batch_mutate(outerMap, CL);

		LOG.debug("Done inserting Phoenix Zoo.");
	}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
private void insertPOIEmpireState() throws Exception {

		Map<ByteBuffer, Map<String, List<Mutation>>> outerMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();

		List<Mutation> columnsToAdd = new ArrayList<Mutation>();

		long ts = System.nanoTime();
		String esbName = "Empire State Building";
		Column descCol = new Column(bytes("desc"));
		Column phoneCol = new Column(bytes("phone"));

		List<Column> esbCols = new ArrayList<Column>();
		esbCols.add(descCol);
		esbCols.add(phoneCol);

		Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

		Mutation columns = new Mutation();
		ColumnOrSuperColumn descCosc = new ColumnOrSuperColumn();
		SuperColumn waldorfSC = new SuperColumn();
		waldorfSC.name = bytes(WALDORF_NAME);
		waldorfSC.columns = esbCols;

		descCosc.super_column = waldorfSC;
		columns.setColumn_or_supercolumn(descCosc);

		columnsToAdd.add(columns);

		String superCFName = "PointOfInterest";
		ColumnPath cp = new ColumnPath();
		cp.column_family = superCFName;
		cp.setSuper_column(WALDORF_NAME.getBytes());
		cp.setSuper_columnIsSet(true);

		innerMap.put(superCFName, columnsToAdd);
		outerMap.put(bytes(esbName), innerMap);

		client.batch_mutate(outerMap, CL);

		LOG.debug("Done inserting Empire State.");
	}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
public List<POI> findPOIByHotel(String hotel) throws Exception {

		// /query
		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange();
		sliceRange.setStart(hotel.getBytes());
		sliceRange.setFinish(hotel.getBytes());
		predicate.setSlice_range(sliceRange);

		// read all columns in the row
		String scFamily = "PointOfInterest";
		ColumnParent parent = new ColumnParent(scFamily);

		KeyRange keyRange = new KeyRange();
		keyRange.start_key = bytes("");
		keyRange.end_key = bytes("");

		List<POI> pois = new ArrayList<POI>();

		// instead of a simple list, we get a map whose keys are row keys
		// and the values the list of columns returned for each
		// only row key + first column are indexed
		Connector cl = new Connector();
		Cassandra.Client client = cl.connect();
		List<KeySlice> slices = client.get_range_slices(parent, predicate, keyRange, CL);

		for (KeySlice slice : slices) {
			List<ColumnOrSuperColumn> cols = slice.columns;

			POI poi = new POI();
			poi.name = new String(ByteBufferUtil.string(slice.key));

			for (ColumnOrSuperColumn cosc : cols) {
				SuperColumn sc = cosc.super_column;

				List<Column> colsInSc = sc.columns;

				for (Column c : colsInSc) {
					String colName = new String(c.name.array(), UTF8);
					if (colName.equals("desc")) {
						poi.desc = new String(c.value.array(), UTF8);
					}
					if (colName.equals("phone")) {
						poi.phone = new String(c.value.array(), UTF8);
					}
				}

				LOG.debug("Found something neat nearby: " + poi.name + ". \nDesc: " + poi.desc + ". \nPhone: "
						+ poi.phone);
				pois.add(poi);
			}
		}

		cl.close();
		return pois;
	}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
public void run(final ThriftClient client) throws IOException
{
    final SlicePredicate predicate = slicePredicate();
    final ByteBuffer key = getKey();
    final List<ByteBuffer> expect = state.rowGen.isDeterministic() ? generateColumnValues(key) : null;
    for (final ColumnParent parent : state.columnParents)
    {
        timeWithRetry(new RunOp()
        {
            @Override
            public boolean run() throws Exception
            {
                List<ColumnOrSuperColumn> row = client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel);
                if (expect == null)
                    return !row.isEmpty();
                if (row == null)
                    return false;
                if (!state.settings.columns.useSuperColumns)
                {
                    if (row.size() != expect.size())
                        return false;
                    for (int i = 0 ; i < row.size() ; i++)
                        if (!row.get(i).getColumn().bufferForValue().equals(expect.get(i)))
                            return false;
                }
                else
                {
                    for (ColumnOrSuperColumn col : row)
                    {
                        SuperColumn superColumn = col.getSuper_column();
                        if (superColumn.getColumns().size() != expect.size())
                            return false;
                        for (int i = 0 ; i < expect.size() ; i++)
                            if (!superColumn.getColumns().get(i).bufferForValue().equals(expect.get(i)))
                                return false;
                    }
                }
                return true;
            }

            @Override
            public String key()
            {
                return new String(key.array());
            }

            @Override
            public int keyCount()
            {
                return 1;
            }
        });
    }
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
@Override
public void write(String keySpace, CassandraProxyClient client, JobConf jc) throws IOException {
    ConsistencyLevel flevel = getConsistencyLevel(jc);
    int batchMutation = getBatchMutationSize(jc);
    Map<ByteBuffer, Map<String, List<Mutation>>> mutation_map = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
    Map<String, List<Mutation>> maps = new HashMap<String, List<Mutation>>();

    int count = 0;
    for (CassandraPut c : subColumns) {
        List<Column> columns = new ArrayList<Column>();
        for (CassandraColumn col : c.getColumns()) {
            Column cassCol = new Column();
            cassCol.setValue(col.getValue());
            cassCol.setTimestamp(col.getTimeStamp());
            cassCol.setName(col.getColumn());
            columns.add(cassCol);

            ColumnOrSuperColumn thisSuperCol = new ColumnOrSuperColumn();
            thisSuperCol.setSuper_column(new SuperColumn(c.getKey(), columns));

            Mutation mutation = new Mutation();
            mutation.setColumn_or_supercolumn(thisSuperCol);

            List<Mutation> mutList = maps.get(col.getColumnFamily());
            if (mutList == null) {
                mutList = new ArrayList<Mutation>();
                maps.put(col.getColumnFamily(), mutList);
            }

            mutList.add(mutation);

            count++;

            if (count == batchMutation) {
                mutation_map.put(key, maps);

                commitChanges(keySpace, client, flevel, mutation_map);

                //reset mutation map, maps and count;
                mutation_map = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
                maps = new HashMap<String, List<Mutation>>();
                count = 0;
            }

        }
    }

    if (count > 0) {
        mutation_map.put(key, maps);
        commitChanges(keySpace, client, flevel, mutation_map);
    }
}
 

import org.apache.cassandra.thrift.SuperColumn; //导入依赖的package包/类
@Override
public void write(String keySpace, CassandraProxyClient client, JobConf jc) throws IOException {
  ConsistencyLevel flevel = getConsistencyLevel(jc);
  int batchMutation = getBatchMutationSize(jc);
  Map<ByteBuffer,Map<String,List<Mutation>>> mutation_map = new HashMap<ByteBuffer,Map<String,List<Mutation>>>();
  Map<String, List<Mutation>> maps = new HashMap<String, List<Mutation>>();

  int count = 0;
  for (CassandraPut c : subColumns) {
    List<Column> columns = new ArrayList<Column>();
    for (CassandraColumn col : c.getColumns()) {
      Column cassCol = new Column();
      cassCol.setValue(col.getValue());
      cassCol.setTimestamp(col.getTimeStamp());
      cassCol.setName(col.getColumn());
      columns.add(cassCol);

      ColumnOrSuperColumn thisSuperCol = new ColumnOrSuperColumn();
      thisSuperCol.setSuper_column(new SuperColumn(c.getKey(), columns));

      Mutation mutation = new Mutation();
      mutation.setColumn_or_supercolumn(thisSuperCol);

      List<Mutation> mutList = maps.get(col.getColumnFamily());
      if (mutList == null) {
        mutList = new ArrayList<Mutation>();
        maps.put(col.getColumnFamily(), mutList);
      }

      mutList.add(mutation);

      count ++;

      if (count == batchMutation) {
        mutation_map.put(key, maps);

        commitChanges(keySpace, client, flevel, mutation_map);

        //reset mutation map, maps and count;
        mutation_map = new HashMap<ByteBuffer,Map<String,List<Mutation>>>();
        maps = new HashMap<String, List<Mutation>>();
        count = 0;
      }

    }
  }

  if(count > 0) {
    mutation_map.put(key, maps);
    commitChanges(keySpace, client, flevel, mutation_map);
  }
}