cassandra学习笔记5--使用Binary Memtable将大量数据导入Cassandra

约定不等于承诺〃 2022-05-15 05:14 262阅读 0赞

在[Cassandra的客户端][Cassandra] 中，我们谈到了如何使用Thrift API以及更加高级的封装（Hector）如果将数据导入到到Cassandra中，但是在导入大量数据的时候这会遇到很多的问题，比如插入失败，超时等等问题。为了解决这个问题，我们可以尝试使用Binary Memtable。

在Cassandra的wiki上，对Binary Memtable的描述如下：

> Binary Memtable is the name of Cassandra's bulk-load interface. It avoids several kinds of overhead associated with the normal Thrift API:
> 
>  *  Converting to Thrift from the internal structures and back
>  *  Routing (copying) from a coordinator node to the replica nodes
>  *  Writing to the commitlog
>  *  Serializing the internal structures to on-disk format
> 
> The tradeoff you make is that it is considerably less convenient to use than Thrift:
> 
>  *  You must use the [StorageProxy][] API, only available as Java code
>  *  You must pre-serialize the rows yourself
>  *  The rows you send are not live for querying until a flush occurs (either normally because the Binary Memtable fills up, or because you request one with nodetool )
>  *  You must write an entire row at once

具体的实现我们可以参考[https://svn.apache.org/repos/asf/cassandra/trunk/contrib/bmt\_example/][https_svn.apache.org_repos_asf_cassandra_trunk_contrib_bmt_example] .但是这个示例结合了Hadoop，如何没有这个环境，也可以参考我的测试代码：

package
     com.alibaba.dw.thrift.client;

import
     java.io.IOException;

import
     java.io.UnsupportedEncodingException;

import
     org.apache.cassandra.config.DatabaseDescriptor;

import
     org.apache.cassandra.db.Column;

import
     org.apache.cassandra.db.ColumnFamily;

import
     org.apache.cassandra.db.RowMutation;

import
     org.apache.cassandra.db.filter.QueryPath;

import
     org.apache.cassandra.io.util.DataOutputBuffer;

import
     java.net.InetAddress;

import
     org.apache.cassandra.net.Message;

import
     org.apache.cassandra.net.MessagingService;

import
     org.apache.cassandra.service.StorageService;

/**

* TODO Comment of BinaryMemtableTest

* @author aaron.guop

public
     class
     BinaryMemtableTest {

/**

* @param args

* @throws IOException

* @throws InterruptedException

public
     static
     void
     main(String[] args) throws
     IOException, InterruptedException {

System.setProperty("storage-config"
    , "D:\\apache-cassandra-0.6.1\\conf"
    );

StorageService.instance.initClient();

while
     (StorageService.instance.getNaturalEndpoints("Keyspace1"
    , "bmt"
    ).isEmpty()) {

Thread.sleep(1 * 1000);

}

doInsert();

StorageService.instance.stopClient();

}

/**

* @throws UnsupportedEncodingException

private
     static
     void
     doInsert() throws
     UnsupportedEncodingException {

String keyspace = "Keyspace1"
    ;

String cfName = "Member"
    ;

String memberID = "bmt"
    ;

/* Create a column family */

ColumnFamily columnFamily = ColumnFamily.create(keyspace, cfName);

//while (values.hasNext()) {

String SuperColumnName = "SuperColumnName"
    ;

String ColumnName = "ColumnName"
    ;

String ColumnValue = "ColumnValue"
    ;

long
     timestamp = 0;

columnFamily.addColumn(new
     QueryPath(cfName, SuperColumnName.getBytes("UTF-8"
    ), ColumnName

.getBytes("UTF-8"
    )), ColumnValue.getBytes(), timestamp);

//}

/* Get serialized message to send to cluster */

Message message = createMessage(keyspace, memberID, cfName, columnFamily);

for
     (InetAddress endpoint : StorageService.instance.getNaturalEndpoints(keyspace, memberID)) {

/* Send message to end point */

MessagingService.instance.sendOneWay(message, endpoint);

System.out.println("Send message to "
     + endpoint.toString());

}

public
     static
     Message createMessage(String Keyspace, String Key, String CFName,

ColumnFamily columnFamile) {

DataOutputBuffer bufOut = new
     DataOutputBuffer();

Column column;

* Get the first column family from list, this is just to get past

* validation

ColumnFamily baseColumnFamily = new
     ColumnFamily(CFName, "Standard"
    , DatabaseDescriptor

.getComparator(Keyspace, CFName), DatabaseDescriptor.getSubComparator(Keyspace,

CFName));

bufOut.reset();

try
     {

ColumnFamily.serializer().serializeWithIndexes(columnFamile, bufOut);

byte
    [] data = new
     byte
    [bufOut.getLength()];

System.arraycopy(bufOut.getData(), 0, data, 0, bufOut.getLength());

column = new
     Column(columnFamile.name().getBytes("UTF-8"
    ), data, 0, false);

baseColumnFamily.addColumn(column);

} catch
     (IOException e) {

throw
     new
     RuntimeException(e);

}

RowMutation rm = new
     RowMutation(Keyspace, Key);

rm.add(baseColumnFamily);

try
     {

/* Make message */

return
     rm.makeRowMutationMessage(StorageService.Verb.BINARY);

} catch
     (IOException e) {

throw
     new
     RuntimeException(e);

}

在运行这个代码之前，我们还必须注意以下事项：

1 运行这个程序的机器不能运行Cassandra Server。

2 在storage-conf.xml的seed中指定其他Cassandra Server的地址。

3 使用nodetool 的getcachecapacity得到当前的压缩设置，记录下来，供完成导入后恢复使用。

4 使用nodetool 的setcachecapacity设置当前的压缩设置为0，即取消压缩功能。

5 运行导入程序。

6 使用nodetool 的flush。

7 使用nodetool 的setcompactionthreshold设置回之前在第三步获得的值。

经过这些操作后，数据就导入到我们的Cassandra中了

[Cassandra]: http://www.cnblogs.com/gpcuster/archive/2010/03/23/1692794.html
[StorageProxy]: http://wiki.apache.org/cassandra/StorageProxy
[https_svn.apache.org_repos_asf_cassandra_trunk_contrib_bmt_example]: https://svn.apache.org/repos/asf/cassandra/trunk/contrib/bmt_example/