Apache Cassandra Astyanax

04 Temmuz 2012 Hakan İlter Nosql, 8

Astyanax, Java ile geliştirilmiş bir Apache Cassandra istemci kütüphanesidir. Netflix tarafından geliştirilmiş ve halen kullanılmaya devam etmektedir. Bu sebeple sık sık yeni geliştirmeler yapılmakta ve problemleri hızlıca çözülmektedir.

Astyanax’ın öne çıkan özelliklerini Thrift API’si ile uğraşmadan daha yüksek seviyeden, arayüz (interface) tabanlı, nesne yönelimli tasarıma sahip bir yapı ile Cassandra erişimi, istemci tarafında hata kontrolü ve otomatik olarak yeniden deneme (retry) mekanizması ve bağlantı havuzu (connection pool) desteği olarak sayabiliriz. Bunun dışında yeterli bir dökümantasyona sahip olduğunu ve ilham aldığı Hector projesinden çok daha düzgün bir tasarıma sahip olduğunu söyleyebiliriz. Sık sık güncelleniyor olması da gerçekten önemli bir ayrıntı.

Devam etmeden önce, eğer Cassandra kurulumu ile ilgili bilgi almak isterseniz Apache Cassandra Kurulumu yazımıza göz atabilirsiniz.

Astyanax’ı maven desteği ile kullanmak için aşağıdaki paketlere ihtiyacımız var:

<dependency>
  <groupId>com.netflix.astyanax</groupId>
  <artifactId>astyanax</artifactId>
  <version>1.0.1</version>
</dependency>
<dependency>
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  <version>11.0.2</version>
</dependency>

 

Paketleri ekledikten sonra örnek amacıyla Cassandra’daki verileri Map objesi ile taşıyan çok basit bir DAO sınıfı yazalım:

import java.util.HashMap;
import java.util.Map;

import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.ColumnListMutation;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.OperationResult;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.Column;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;

public class BasicCassandraAstyanaxDao 
{
	private static final StringSerializer SS = StringSerializer.get();

	private String hosts;
	private String clusterName;
	private String keyspaceName;

	private AstyanaxContext<Keyspace> context;
	private Keyspace keyspace;
	private ColumnFamily<String, String> columnFamily;

	public BasicCassandraAstyanaxDao(String hosts, String clusterName, String keyspaceName, String columnFamilyName)
	{
		this.hosts = hosts;
		this.clusterName = clusterName;
		this.keyspaceName = keyspaceName;
		this.columnFamily = ColumnFamily.newColumnFamily(columnFamilyName, SS, SS);
	}

	public Keyspace getKeyspace()
	{
		if (keyspace == null) {
			context = new AstyanaxContext.Builder()
				.forCluster(clusterName)
				.forKeyspace(keyspaceName)
				.withAstyanaxConfiguration(new AstyanaxConfigurationImpl()      
				.setDiscoveryType(NodeDiscoveryType.NONE)
			)
				.withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl(keyspaceName + "Pool")
				.setMaxConnsPerHost(1)
				.setSeeds(hosts)
			)
			.withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
			.buildKeyspace(ThriftFamilyFactory.getInstance());
			context.start();
			keyspace = context.getEntity();
		}
		return keyspace;
	}

	public void put(String key, Map<String, String> map) throws ConnectionException 
	{
		MutationBatch mutation = getKeyspace().prepareMutationBatch();
		ColumnListMutation<String> clm = mutation.withRow(columnFamily, key);
		for (Map.Entry<String, String> entry : map.entrySet()) {
			if (entry.getValue() != null) {
				clm.putColumn(entry.getKey(), entry.getValue(), null);	
			}
		}
		mutation.execute();
	}

	public Map<String, String> get(String key) throws ConnectionException
	{
		Map<String, String> map = null;

		OperationResult<ColumnList<String>> result = getKeyspace().prepareQuery(columnFamily).getKey(key).execute();
		if (result != null) {
			map = new HashMap<String, String>();
			for (Column<String> column : result.getResult()) {
				map.put(column.getName(), column.getStringValue());
			}							
		}

		if (map != null && map.size() == 0) {
			map = null;
		}
		return map;
	}

	public void delete(String key) throws ConnectionException 
	{
		MutationBatch mutation = getKeyspace().prepareMutationBatch();
		mutation.withRow(columnFamily, key).delete();
		mutation.execute();
	}

	public void destroy()
	{
		if (context != null) {
			context.shutdown();
			context = null;
			keyspace = null;
		}
	}

 

Bu sınıfı kullanan basit bir main metodu yazalım:

	public static void main(String[] args) throws ConnectionException 
	{
		final String HOSTS = "127.0.0.1:9160";
		final String CLUSTER_NAME = "Test";
		final String KEYSPACE_NAME = "Test";
		final String COLUMN_FAMILY_NAME = "Test1";

		// create dao instance 
		BasicCassandraAstyanaxDao dao = new BasicCassandraAstyanaxDao(HOSTS, CLUSTER_NAME, KEYSPACE_NAME, COLUMN_FAMILY_NAME);

		// save
		Map<String, String> saved = new HashMap<String, String>();
		saved.put("name", "Cassandra");
		saved.put("email", "cassandra@apache.org");
		dao.put("1", saved);
		System.out.println("saved object: " + saved);

		// get
		Map<String, String> loaded = dao.get("1");
		Assert.assertNotNull(loaded);
		Assert.assertEquals(saved.get("name"), loaded.get("name"));
		Assert.assertEquals(saved.get("email"), loaded.get("email"));
		System.out.println("loaded object: " + loaded);

		// delete
		dao.delete("1");
		Map<String, String> deleted = dao.get("1");
		Assert.assertNull(deleted);
		System.out.println("deleted object: " + deleted);

		// destroy dao instance
		dao.destroy();
	}

 

Yazdığımız kodu çalıştırmadan önce Cassandra’yı çalıştırdıktan sonra örnekte kullanacağımız Keyspace ve Column Family nesnelerini oluşturmak için cassandra-cli kullanalım:

$ /usr/java/cassandra/bin/cassandra-cli
Connected to: "Test Cluster" on 127.0.0.1/9160
Welcome to Cassandra CLI version 1.1.1

Type 'help;' or '?' for help.
Type 'quit;' or 'exit;' to quit.

[default@unknown] create keyspace Test;
174196b6-d45d-34e5-b051-659f933c2012
Waiting for schema agreement...
... schemas agree across the cluster
[default@unknown] use Test;
Authenticated to keyspace: Test
[default@Test] create column family Test1 with comparator = UTF8Type;
bc69772c-169e-3d20-bb86-fbb478d8d771
Waiting for schema agreement...
... schemas agree across the cluster
[default@Test]

 

Artık test kodumuzu çalıştırabiliriz, test kodumuzu çalıştığında Cassandra’ya bağlanıyor, email ve name alanlarını içeren anahtar değeri 1 olan kaydı yapıyor, daha sonra bu kaydı Cassandra’dan çekiyor. Çektiği kaydı kaydettiği kayıt ile karşılaştırdıktan sonra silip, sildiğinden de emin oluyor.

2012-07-04 23:02:31,731 INFO  [com.netflix.astyanax.connectionpool.impl.ConnectionPoolMBeanManager] Registering mbean: com.netflix.MonitoredResources:type=ASTYANAX,name=TestPool,ServiceType=connectionpool
saved object: {email=cassandra@apache.org, name=Cassandra}
loaded object: {email=cassandra@apache.org, name=Cassandra}
deleted object: null

Bu basit örneğin işinize yarayacağını umarım, daha detaylı bilgiye Astyanax dökümantasyonundan ulaşabilirsiniz.

Astyanax, Cassandra, NoSQLDB

Cloudera CDH3 Hadoop Kurulumu Hadoop MapReduce Örnek Uygulama