Cassandra事始め

事前準備

インストールは、Cassandra Wiki を参照
※ Cassandra のバージョンは1.2.3を使用

keyspaceの作成

cassandra-cliでCassandraに接続していることが前提。
また、keyspace名は、仮にKeyspace1として記述する。

作成

create keyspace Keyspace1;を実行する。

ColumnFamilyの作成

ここでは、作成するColumnFamily名はStandard1として記述する。

keyspaceの選択

use Keyspace1を実行する。

作成

create column family Standard1;を実行する。
※ オプションは調べておらず、未指定で動作したため、特に指定していない。

アプリからの接続

ここでは、Cassandra WikiのThriftExamplesのJava項を参考に(言語自体の勉強のため)ScalaとGradle、TestNGを使って、簡単なClientを作ってみる。
※ Scalaの構文とGradleの使い方、TestNGの使い方および、プロジェクトの作成の説明は割愛する

build.gradle

使用したbuild.gradleは以下のとおり。
```build.gradle
apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'scala'

sourceCompatibility = 1.7
version = '1.0'
jar {
manifest {
attributes 'Implementation-Title': 'Gradle Quickstart', 'Implementation-Version': version
}
}

repositories {
mavenCentral()
}

dependencies {
compile 'org.apache.thrift:libthrift:0.9.0'
compile 'org.apache.cassandra:cassandra-thrift:1.2.3'
testCompile 'org.testng:testng:6.8'
}

test {
systemProperties 'property': 'value'
}

uploadArchives {
repositories {
flatDir {
dirs 'repos'
}
}
}
```

クライアントコード

?(詳細未調査)と記述されている部分に関しては、詳細を調べておらず、よく分からないため、コメント内容は無視してください。

CassandraClient.scala

package learning.client.cassandra

import org.apache.cassandra.thrift.Cassandra
import org.apache.cassandra.thrift.Column
import org.apache.cassandra.thrift.ColumnParent
import org.apache.cassandra.thrift.ConsistencyLevel
import org.apache.thrift.protocol.TBinaryProtocol
import org.apache.thrift.transport.TFramedTransport
import org.apache.thrift.transport.TSocket
import java.nio.ByteBuffer
import java.io.Closeable
import org.apache.cassandra.thrift.ColumnOrSuperColumn
import org.apache.cassandra.thrift.ColumnPath
import org.apache.cassandra.thrift.SliceRange
import org.apache.cassandra.thrift.SlicePredicate

/**
 * @author PoaD
 *
 */
class CassandraClient(hostname : String, port : Int, keyspace : String) extends Closeable {
    // おまじない（接続するための設定）
    private val transport = new TFramedTransport(new TSocket(hostname, port))
    private val protocol = new TBinaryProtocol(transport)
    // 接続
    transport.open()
    // Clientオブジェクトの生成
    private val client = new Cassandra.Client(protocol)
    // 使用するkeyspaceの設定
    client.set_keyspace(keyspace)

    /**
     * valueで指定された値をkeyおよび、parentに該当するcolumnとして書き込みます。
     *
     * @param key insertするデータのキー
     * @param parent insertする先のColumnFamily
     * @param value insertする値
     * @param level データ保持担当Nodeを決定するための値{@link http://lunarium.info/arc/index.php/Cassandra%E3%81%AE%E6%A7%8B%E9%80%A0#Data.E6.93.8D.E4.BD.9C.E3.81.AEAlgorithm} を参照
     */
    def insert(key : String, parent : ColumnParent, name : ByteBuffer, value : ByteBuffer, level : ConsistencyLevel) : Unit = {
        val timestamp = System.currentTimeMillis()
        val column = new Column()
        column.setName(name)
        column.setValue(value)
        column.setTimestamp(timestamp)
        client.insert(ByteBuffer.wrap(key.getBytes("UTF-8")), parent, column, level)
    }

    /**
     * keyおよび、parentに該当するcolumnの一覧を返します。
     *
     * @param key 取得するデータのキー
     * @param parent 取得する元のColumnFamily
     * @param start 取得するデータの範囲の開始位置?(詳細未調査)
     * @param end 取得するデータの範囲の終了位置?(詳細未調査)
     * @param reversed 並び順を反転?(詳細未調査)
     * @param count 取得するデータの最大数?(詳細未調査)
     * @param level データ保持担当Nodeを決定するための値{@link http://lunarium.info/arc/index.php/Cassandra%E3%81%AE%E6%A7%8B%E9%80%A0#Data.E6.93.8D.E4.BD.9C.E3.81.AEAlgorithm} を参照
     * @return keyおよび、parentに該当するcolumnの一覧
     */
    def getClumns(key : String, parent : ColumnParent, start : ByteBuffer, end : ByteBuffer, reversed : Boolean, count : Int, level : ConsistencyLevel) : java.util.List[ColumnOrSuperColumn] = {
        val predicate = new SlicePredicate()
        predicate.setSlice_range(new SliceRange(start, end, reversed, count))
        client.get_slice(ByteBuffer.wrap(key.getBytes("UTF-8")), parent, predicate, level)
    }

    /**
     * クライアントを終了します。
     */
    def close = {
        transport.flush()
        transport.close()
    }
}

テストコード

ざっとこんな感じ。書いて読み込んで、内容が同じかチェックして終わり。

CassandraClientTest.scala

package learning.client.cassandra

import org.testng.annotations.Test
import org.testng.annotations.BeforeClass
import org.apache.cassandra.thrift.ColumnPath
import org.apache.cassandra.thrift.ColumnParent
import org.apache.cassandra.thrift.ConsistencyLevel
import java.nio.ByteBuffer
import org.apache.cassandra.thrift.SlicePredicate
import org.apache.cassandra.thrift.SliceRange
import org.apache.cassandra.thrift.ColumnOrSuperColumn
import org.testng.Assert

import scala.collection.convert.DecorateAsJava

/**
 * @author PoaD
 *
 */
class CassandraClientTest {

    @Test
    def test() = {
        // テスト対象
        val target = new CassandraClient("localhost", 9160, "Keyspace1")

        val key = "1"
        val name = "name"
        val value = "test value"
        val parent = new ColumnParent("Standard1") // Column Familyの指定
        // 書き込み
        target.insert(
            key,
            parent,
            ByteBuffer.wrap(name.getBytes("UTF-8")),
            ByteBuffer.wrap(value.getBytes("UTF-8")),
            ConsistencyLevel.ALL)

        // 書き込んだ結果の取得
        val columns : java.util.List[ColumnOrSuperColumn] = target.getClumns(
            key,
            parent,
            ByteBuffer.wrap(new Array[Byte](0)),
            ByteBuffer.wrap(new Array[Byte](0)),
            false,
            100,
            ConsistencyLevel.ALL)
        // 取得内容の検証
        Assert.assertEquals(columns.size(), 1)
        Assert.assertEquals(columns.get(0).getColumn().getName(), name.getBytes("UTF-8"))
        Assert.assertEquals(columns.get(0).getColumn().getValue(), value.getBytes("UTF-8"))
        target.close // clientの終了
    }
}