Java 类org.apache.cassandra.thrift.SuperColumn 实例源码

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;
}

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;
}

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;
}

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;
}

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.");
}

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.");
    }

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;
}

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;
}

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;
}

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;
}

/**
 * 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;
}

/**
 * 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;
}

/**
 * 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;
}

/**
 * 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;
}

/**
 * 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;
}

/**
 * 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;
}

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.");
    }

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.");
    }

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;
    }

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;
            }
        });
    }
}

@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);
    }
}

@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);
  }
}

/**
 * Insert into IpUser Column Family
 */
public boolean InsertIpUser(String ip, String name, String time, int times) {
    System.out.println("------------InsertIpUser--------------");
    String IP_KEY = ip;
    String NAME_SUPER_KEY = name;
    String TIMES = String.valueOf(times);
    String LAST = this.tc.NormalTime(time);

    String COLUMN_NAME_TIMES = "times";
    String COLUMN_NAME_LAST  = "last";

    String COLUMN_FAMILY_NAME = "IpUser";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnTimes = new Column();
        idColumnTimes.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_TIMES));
        idColumnTimes.setValue(this.cassandraUtil.toByteBuffer(TIMES));
        idColumnTimes.setTimestamp(timeStamp);

        Column idColumnLast = new Column();
        idColumnLast.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_LAST));
        idColumnLast.setValue(this.cassandraUtil.toByteBuffer(LAST));
        idColumnLast.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(NAME_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY_NAME;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(NAME_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY_NAME, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(IP_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.ONE);

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

/**
 * Insert into UserIp Column Family
 */
public boolean InsertUserIp(String ip, String name, String time, int times) {
    System.out.println("------------InsertUserIp--------------");
    String NAME_KEY = name;
    String IP_SUPER_KEY = ip;
    String TIMES = String.valueOf(times);
    String LAST = this.tc.NormalTime(time);

    String COLUMN_NAME_TIMES = "times";
    String COLUMN_NAME_LAST  = "last";

    String COLUMN_FAMILY_NAME = "UserIp";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnTimes = new Column();
        idColumnTimes.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_TIMES));
        idColumnTimes.setValue(this.cassandraUtil.toByteBuffer(TIMES));
        idColumnTimes.setTimestamp(timeStamp);

        Column idColumnLast = new Column();
        idColumnLast.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_LAST));
        idColumnLast.setValue(this.cassandraUtil.toByteBuffer(LAST));
        idColumnLast.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(IP_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY_NAME;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(IP_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY_NAME, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(NAME_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.QUORUM);

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

/**
 * Insert into BookIp CF
 */
public boolean InsertIntoBookIp(String ip, String bookid, String last, int times) {
    System.out.println("------------InsertIntoBookIp--------------");
    String BOOKID_KEY = bookid;
    String IP_SUPER_KEY = ip;
    String TIMES = String.valueOf(times);
    String LAST = this.tc.NormalTime(last);

    String COLUMN_NAME_TIMES = "times";
    String COLUMN_NAME_LAST  = "last";

    String COLUMN_FAMILY_NAME = "BookIp";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnTimes = new Column();
        idColumnTimes.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_TIMES));
        idColumnTimes.setValue(this.cassandraUtil.toByteBuffer(TIMES));
        idColumnTimes.setTimestamp(timeStamp);

        Column idColumnLast = new Column();
        idColumnLast.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_LAST));
        idColumnLast.setValue(this.cassandraUtil.toByteBuffer(LAST));
        idColumnLast.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(IP_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY_NAME;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(IP_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY_NAME, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(BOOKID_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.QUORUM);

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

/**
 * Insert into BookUser CF
 */
public boolean InsertIntoBookUser(String user, String bookid, String last, int times){
    System.out.println("------------InsertIntoBookUser--------------");
    String BOOKID_KEY = bookid;
    String USER_SUPER_KEY = user;
    String TIMES = String.valueOf(times);
    String LAST = this.tc.NormalTime(last);

    String COLUMN_NAME_TIMES = "times";
    String COLUMN_NAME_LAST  = "last";

    String COLUMN_FAMILY_NAME = "BookUser";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnTimes = new Column();
        idColumnTimes.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_TIMES));
        idColumnTimes.setValue(this.cassandraUtil.toByteBuffer(TIMES));
        idColumnTimes.setTimestamp(timeStamp);

        Column idColumnLast = new Column();
        idColumnLast.setName(this.cassandraUtil.toByteBuffer(COLUMN_NAME_LAST));
        idColumnLast.setValue(this.cassandraUtil.toByteBuffer(LAST));
        idColumnLast.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(USER_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY_NAME;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(USER_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY_NAME, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(BOOKID_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.QUORUM);

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

/**
 * Insert into IpBookPage CF
 */
public boolean InsertIntoIpBookPage(String ip, String book, String page, int times, String time) {
    System.out.println("------------InsertIntoIpBookPage--------------");
    String IP_KEY = ip;
    String BOOK_SUPER_KEY = book;

    String PAGE_TITLE = page;
    String PAGE_VALUE = this.tc.NormalTime(time);

    String TIMES_TITLE = "times";
    String TIMES_VALUE = String.valueOf(times);

    String COLUMN_FAMILY = "IpBookPage";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnTimes = new Column();
        idColumnTimes.setName(this.cassandraUtil.toByteBuffer(TIMES_TITLE));
        idColumnTimes.setValue(this.cassandraUtil.toByteBuffer(TIMES_VALUE));
        idColumnTimes.setTimestamp(timeStamp);

        Column idColumnPage = new Column();
        idColumnPage.setName(this.cassandraUtil.toByteBuffer(PAGE_TITLE));
        idColumnPage.setValue(this.cassandraUtil.toByteBuffer(PAGE_VALUE));
        idColumnPage.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(BOOK_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(BOOK_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(IP_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.ONE);

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

/**
 * Insert into UserBookPage CF
 */
public boolean InsertIntoUserBookPage(String user, String book, String page, int times, String time) {
    System.out.println("------------InsertIntoUserBookPage--------------");
    String USER_KEY = user;
    String BOOK_SUPER_KEY = book;

    String PAGE_TITLE = page;
    String PAGE_VALUE = this.tc.NormalTime(time);

    String TIMES_TITLE = "times";
    String TIMES_VALUE = String.valueOf(times);

    String COLUMN_FAMILY = "UserBookPage";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnTimes = new Column();
        idColumnTimes.setName(this.cassandraUtil.toByteBuffer(TIMES_TITLE));
        idColumnTimes.setValue(this.cassandraUtil.toByteBuffer(TIMES_VALUE));
        idColumnTimes.setTimestamp(timeStamp);

        Column idColumnPage = new Column();
        idColumnPage.setName(this.cassandraUtil.toByteBuffer(PAGE_TITLE));
        idColumnPage.setValue(this.cassandraUtil.toByteBuffer(PAGE_VALUE));
        idColumnPage.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(BOOK_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(BOOK_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(USER_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.QUORUM);

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

/**
 * Insert into IpQueryBook CF
 */
public boolean InsertIpQueryBook(String ip, String superUuid, String query,String book, String time) {
    System.out.println("------------InsertIpQueryBook--------------");
    String IP_KEY = ip;
    String UUID_SUPER_KEY = superUuid;

    String QUERY_TITLE = "query";
    String QUERY_VALUE = query;

    String BOOK_TITLE = book;
    String BOOK_VALUE = time;

    String COLUMN_FAMILY = "IpQueryBook";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnQuery = new Column();
        idColumnQuery.setName(this.cassandraUtil.toByteBuffer(QUERY_TITLE));
        idColumnQuery.setValue(this.cassandraUtil.toByteBuffer(QUERY_VALUE));
        idColumnQuery.setTimestamp(timeStamp);

        Column idColumnBook = new Column();
        idColumnBook.setName(this.cassandraUtil.toByteBuffer(BOOK_TITLE));
        idColumnBook.setValue(this.cassandraUtil.toByteBuffer(BOOK_VALUE));
        idColumnBook.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(UUID_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(UUID_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(IP_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.QUORUM);         

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

/**
 * Insert into UserQueryBook CF
 */
public boolean InsertUserQueryBook(String name, String superUuid, String query, String book, String time) {
    System.out.println("------------InsertUserQueryBook--------------");
    String USER_KEY = name;
    String UUID_SUPER_KEY = superUuid;

    String QUERY_TITLE = "query";
    String QUERY_VALUE = query;

    String BOOK_TITLE = book;
    String BOOK_VALUE = time;

    String COLUMN_FAMILY = "UserQueryBook";

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

        long timeStamp = System.currentTimeMillis();

        // Construct Column
        Column idColumnQuery = new Column();
        idColumnQuery.setName(this.cassandraUtil.toByteBuffer(QUERY_TITLE));
        idColumnQuery.setValue(this.cassandraUtil.toByteBuffer(QUERY_VALUE));
        idColumnQuery.setTimestamp(timeStamp);

        Column idColumnBook = new Column();
        idColumnBook.setName(this.cassandraUtil.toByteBuffer(BOOK_TITLE));
        idColumnBook.setValue(this.cassandraUtil.toByteBuffer(BOOK_VALUE));
        idColumnBook.setTimestamp(timeStamp);

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

        // Map Key-super_key-column
        Map<String, List<Mutation>> innerMap = new HashMap<String, List<Mutation>>();

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = this.cassandraUtil.toByteBuffer(UUID_SUPER_KEY);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = COLUMN_FAMILY;
        columnPath.setSuper_column(this.cassandraUtil.toByteBuffer(UUID_SUPER_KEY));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(COLUMN_FAMILY, columnToAdd);

        // Insert Operator
        outerMap.put(this.cassandraUtil.toByteBuffer(USER_KEY), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.QUORUM);         

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

public void superStore(String str) {
    String key_name = "Yanfei";
    String super_key_name = "100300";
    String ColumnFamilyName = "Super";
    String[] strList = str.split(" ");

    try {

        TTransport tr = new TFramedTransport(new TSocket("10.15.61.111", 9160));
        TProtocol proto = new TBinaryProtocol(tr);
        Cassandra.Client client = new Cassandra.Client(proto);

        tr.open();

        client.set_keyspace("Keyspace2");

        long timeStamp = System.currentTimeMillis();

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

        // 
        Column idColumnIp = new Column();
        idColumnIp.setName(toByteBuffer("ip"));
        idColumnIp.setValue(toByteBuffer(strList[0]));
        idColumnIp.setTimestamp(timeStamp);

        Column idColumnBookid = new Column();
        idColumnBookid.setName(toByteBuffer("bookid"));
        idColumnBookid.setValue(toByteBuffer(strList[1]));
        idColumnBookid.setTimestamp(timeStamp);

        Column idColumnPageid = new Column();
        idColumnPageid.setName(toByteBuffer("pageid"));
        idColumnPageid.setValue(toByteBuffer(strList[2]));
        idColumnPageid.setTimestamp(timeStamp);

        List<Column> cols = new ArrayList<Column>();
        cols.add(idColumnIp);
        cols.add(idColumnBookid);
        cols.add(idColumnPageid);

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

        Mutation columns = new Mutation();
        ColumnOrSuperColumn columnOrSuperColumn = new ColumnOrSuperColumn();
        SuperColumn superColumn = new SuperColumn();

        superColumn.name = toByteBuffer(super_key_name);
        superColumn.columns = cols;

        columnOrSuperColumn.super_column = superColumn;
        columns.setColumn_or_supercolumn(columnOrSuperColumn);

        columnToAdd.add(columns);

        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = ColumnFamilyName;
        columnPath.setSuper_column(toByteBuffer(super_key_name));
        columnPath.setSuper_columnIsSet(true);

        innerMap.put(ColumnFamilyName, columnToAdd);

        // 
        outerMap.put(toByteBuffer(key_name), innerMap);

        client.batch_mutate(outerMap, ConsistencyLevel.ONE);

        tr.close();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * Super-Column Query
 */
public void superQuery() {
    TTransport tr = new TFramedTransport(new TSocket("10.15.61.111", 9160));
    TProtocol proto = new TBinaryProtocol(tr);
    Cassandra.Client client = new Cassandra.Client(proto);

    try {
        tr.open();
        client.set_keyspace("Keyspace2");

        // read single column
        System.out.println("------------Single---------------");
        ColumnPath columnPath = new ColumnPath();
        columnPath.column_family = "Super";
        columnPath.super_column = toByteBuffer("100900");

        ColumnOrSuperColumn columnOrSuperColumn = client.get(toByteBuffer("Yanfei"), columnPath, ConsistencyLevel.ONE);

        SuperColumn superColumn = columnOrSuperColumn.getSuper_column();

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

        for(Column col : columns) {
            System.out.println(new String(col.getName(), "UTF-8") + " --> " + new String(col.getValue(), "UTF-8"));
        }

        // read entire row
        System.out.println("------------Entire---------------");
        SlicePredicate predicate = new SlicePredicate();//null, new SliceRange(new byte[0], new byte[0], false, 10)
        SliceRange range = new SliceRange();
        range.start = toByteBuffer("100100");
        range.finish = toByteBuffer("100300");
        predicate.setSlice_range(range);

        ColumnParent parent = new ColumnParent();
        parent.column_family = "Super";

        List<ColumnOrSuperColumn> results = client.get_slice(toByteBuffer("Yanfei"), parent, predicate, ConsistencyLevel.ONE);
        for (ColumnOrSuperColumn result : results) {
            SuperColumn superColumn2 = result.super_column;

            List<Column> columns2 = superColumn2.getColumns();

            System.out.println(new String(superColumn2.getName(), "UTF-8"));

            for (Column column : columns2) {
                System.out.println(new String(column.getName(), "UTF-8") + " -> "+ new String(column.getValue(), "UTF-8"));
            }
        }

        tr.close();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * To query from 'BCPRelation'
 * @param bookChapterList
 */
public void QueryFromBCPRelation (List<List<String>> bookChapterList){      
    SlicePredicate predicate = new SlicePredicate();
    SliceRange range = new SliceRange();

    range.setStart(new byte[0]);
    range.setFinish(new byte[0]); 

    predicate.setSlice_range(range);

    ColumnParent parent = new ColumnParent();
    parent.column_family = "BCPRelation";

    try{
        for (int i = 0; i < bookChapterList.size(); ++i) {
            List<String> innerResult = new ArrayList<String>();

            List<String> innerList = bookChapterList.get(i);

            // Deal with a single book info list
            String bookId = innerList.get(0);
            List<String> chapterList = innerList.subList(1, innerList.size());

            innerResult.add(bookId);

            // Query from cassandra
            List<ColumnOrSuperColumn> results = client.get_slice(this.cassandraUtil.toByteBuffer(bookId), parent, predicate,ConsistencyLevel.ONE);

            // Iterator SuperColumn List
            for (ColumnOrSuperColumn result : results) {
                SuperColumn superColumn2 = result.super_column;
                List<Column> columns2 = superColumn2.getColumns();

                // Get detail information of a single chapter
                String chapterLevel = "";
                String chapterTitle = "";

                for (Column column : columns2) {
                    String columnName = new String(column.getName(), "UTF-8");

                    if (columnName.equalsIgnoreCase("ChapterLevel")) {
                        chapterLevel = new String(column.getValue(), "UTF-8");
                    } else if (columnName.equalsIgnoreCase("ChapterLabel")) {
                        chapterTitle = new String(column.getValue(), "UTF-8");
                    }
                }

                // Iterator chapterList insert into map
                for(int j = 0; j < chapterList.size(); ++j){
                    if(chapterLevel.equals(chapterList.get(j))) {
                        innerResult.add(chapterTitle);
                    }
                }
            }

            this.finalResult.add(innerResult);
        }

    }catch(Exception e) {
        LOG.warn("Error when dealing bookChapterList");
        e.printStackTrace();
    }
}