import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
public RangeSliceCommand deserialize(DataInputStream dis, int version) throws IOException
{
    String keyspace = dis.readUTF();
    String column_family = dis.readUTF();

    int scLength = dis.readInt();
    ByteBuffer super_column = null;
    if (scLength > 0)
        super_column = ByteBuffer.wrap(readBuf(scLength, dis));

    TDeserializer dser = new TDeserializer(new TBinaryProtocol.Factory());
    SlicePredicate pred = new SlicePredicate();
    FBUtilities.deserialize(dser, pred, dis);

    AbstractBounds range = AbstractBounds.serializer().deserialize(dis);
    int max_keys = dis.readInt();
    return new RangeSliceCommand(keyspace, column_family, super_column, pred, range, max_keys);
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
private void dropOldKeyspace() throws InvalidRequestException, SchemaDisagreementException, TException {
  TTransport tr = new TFramedTransport(new TSocket("localhost", 9160));
  TProtocol proto = new TBinaryProtocol(tr);
  Cassandra.Client client = new Cassandra.Client(proto);
  tr.open();

  client.system_drop_keyspace(JANUSGRAPH);
  LOGGER.info("DROPPED keyspace janusgraph");
  tr.close();
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
private static Cassandra.Client createConnection(String host, Integer port, boolean framed) throws IOException
{
    TSocket socket = new TSocket(host, port);
    TTransport trans = framed ? new TFramedTransport(socket) : socket;
    try
    {
        trans.open();
    }
    catch (TTransportException e)
    {
        throw new IOException("unable to connect to server", e);
    }
    return new Cassandra.Client(new TBinaryProtocol(trans));
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
public void serialize(RangeSliceCommand sliceCommand, DataOutputStream dos, int version) throws IOException
{
    dos.writeUTF(sliceCommand.keyspace);
    dos.writeUTF(sliceCommand.column_family);
    ByteBuffer sc = sliceCommand.super_column;
    dos.writeInt(sc == null ? 0 : sc.remaining());
    if (sc != null)
        ByteBufferUtil.write(sc, dos);

    TSerializer ser = new TSerializer(new TBinaryProtocol.Factory());
    FBUtilities.serialize(ser, sliceCommand.predicate, dos);
    AbstractBounds.serializer().serialize(sliceCommand.range, dos);
    dos.writeInt(sliceCommand.max_keys);
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
public void serialize(IndexScanCommand o, DataOutput out) throws IOException
{
    out.writeUTF(o.keyspace);
    out.writeUTF(o.column_family);
    TSerializer ser = new TSerializer(new TBinaryProtocol.Factory());
    FBUtilities.serialize(ser, o.index_clause, out);
    FBUtilities.serialize(ser, o.predicate, out);
    AbstractBounds.serializer().serialize(o.range, out);
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
public IndexScanCommand deserialize(DataInput in) throws IOException
{
    String keyspace = in.readUTF();
    String columnFamily = in.readUTF();

    TDeserializer dser = new TDeserializer(new TBinaryProtocol.Factory());
    IndexClause indexClause = new IndexClause();
    FBUtilities.deserialize(dser, indexClause, in);
    SlicePredicate predicate = new SlicePredicate();
    FBUtilities.deserialize(dser, predicate, in);
    AbstractBounds range = AbstractBounds.serializer().deserialize(in);

    return new IndexScanCommand(keyspace, columnFamily, indexClause, predicate, range);
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
public void serialize(RangeSliceCommand sliceCommand, DataOutput dos, int version) throws IOException
{
    dos.writeUTF(sliceCommand.keyspace);
    dos.writeUTF(sliceCommand.column_family);
    ByteBuffer sc = sliceCommand.super_column;
    dos.writeInt(sc == null ? 0 : sc.remaining());
    if (sc != null)
        ByteBufferUtil.write(sc, dos);

    if (version < MessagingService.VERSION_12)
    {
        FBUtilities.serialize(new TSerializer(new TBinaryProtocol.Factory()), asSlicePredicate(sliceCommand.predicate), dos);
    }
    else
    {
        IDiskAtomFilter.Serializer.instance.serialize(sliceCommand.predicate, dos, version);
    }

    if (version >= MessagingService.VERSION_11)
    {
        if (sliceCommand.row_filter == null)
        {
            dos.writeInt(0);
        }
        else
        {
            dos.writeInt(sliceCommand.row_filter.size());
            for (IndexExpression expr : sliceCommand.row_filter)
            {
                if (version < MessagingService.VERSION_12)
                {
                    FBUtilities.serialize(new TSerializer(new TBinaryProtocol.Factory()), expr, dos);
                }
                else
                {
                    ByteBufferUtil.writeWithShortLength(expr.column_name, dos);
                    dos.writeInt(expr.op.getValue());
                    ByteBufferUtil.writeWithShortLength(expr.value, dos);
                }
            }
        }
    }
    AbstractBounds.serializer.serialize(sliceCommand.range, dos, version);
    dos.writeInt(sliceCommand.maxResults);
    if (version >= MessagingService.VERSION_11)
    {
        dos.writeBoolean(sliceCommand.countCQL3Rows);
        dos.writeBoolean(sliceCommand.isPaging);
    }
}
 

import org.apache.cassandra.thrift.TBinaryProtocol; //导入依赖的package包/类
@Override
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException {
  this.split = (ColumnFamilySplit) split;
  Configuration conf = context.getConfiguration();
  predicate = ConfigHelper.getInputSlicePredicate(conf);
  if (!isSliceRangePredicate(predicate)) {
    throw new AssertionError("WideRowsRequire a slice range");
  }


  totalRowCount = ConfigHelper.getInputSplitSize(conf);
  Log.info("total rows = "+totalRowCount);
  batchRowCount = 1;
  rowPageSize = predicate.getSlice_range().getCount();
  startSlicePredicate = predicate.getSlice_range().start;
  cfName = ConfigHelper.getInputColumnFamily(conf);
  consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getReadConsistencyLevel(conf));


  keyspace = ConfigHelper.getInputKeyspace(conf);

  try {
    // only need to connect once
    if (socket != null && socket.isOpen()) {
      return;
    }

    // create connection using thrift
    String location = getLocation();
    socket = new TSocket(location, ConfigHelper.getInputRpcPort(conf));
    TBinaryProtocol binaryProtocol = new TBinaryProtocol(new TFramedTransport(socket));
    client = new Cassandra.Client(binaryProtocol);
    socket.open();

    // log in
    client.set_keyspace(keyspace);
    if (ConfigHelper.getInputKeyspaceUserName(conf) != null) {
      Map<String, String> creds = new HashMap<String, String>();
      creds.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf));
      creds.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getInputKeyspacePassword(conf));
      AuthenticationRequest authRequest = new AuthenticationRequest(creds);
      client.login(authRequest);
    }
  } catch (Exception e) {
    throw new RuntimeException(e);
  }

  iter = new WideRowIterator();
}