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の終了
	}
}