Apache Cassandra par la pratique

Vous êtes sans doute comblés par l'architecture passionnante de Cassandra (réplication peer-to-peer, niveau de consistance, scalabilité, haute disponibilité, etc.). Aucune de ces qualités n'est vraiment nécessaire lorsque l'on souhaite installer une BD "single-node". Dans beaucoup de situations, une base de données installée sur une seule machine suffira. Par contre, si votre besoin est d'avoir une base de données distribuée sur plusieurs nœuds (même des dizaines), Cassandra est un des meilleurs choix.

Dans une base de données relationnelle, l'écriture des données est coûteuse. Ceci est dû à la manière dont les données sont structurées : le SGBDR doit assurer l'intégrité des données dans les différentes tables au moment de chaque écriture. Cassandra a été conçue pour offrir une meilleure optimisation d'écriture de données. La stratégie d'écriture adoptée par Cassandra ainsi que son incompatibilité avec les transactions ACID permettent de donner plus de performance au niveau de l'écriture des données. Une base de données qui fait beaucoup d'opérations d'écriture est plus performante avec Cassandra.

Cassandra est une base de données orientée colonne, elle s'inspire du modèle BigTable de google. Sur chaque ligne on peut insérer un grand nombre de colonnes (jusqu'à deux milliards). Cassandra offre un plus par rapport au modèle BigTable, car elle propose un concept de conteneur de colonnes ou SuperColumns. Le modèle orienté colonne offre une disposition différente du modèle relationnel. Les colonnes sur Cassandra sont dynamiques et peuvent changer d'une ligne à l'autre.

Cassandra ne définit aucune notion d'intégrité référentielle, il n'y a donc pas de jointures. Dans les bases de données relationnelles, vous pouvez spécifier dans une table la clé étrangère qui référence la clé primaire d'un enregistrement d'une autre table. Mais Cassandra ne supporte pas cette règle, donc les opérations telles que le cascading n'existent pas.

Il est souvent recommandé pour les bases de données relationnelles de respecter la normalisation. Mais ce n'est pas conseillé lorsque l'on souhaite augmenter la performance de la BD. C'est souvent le cas d'entreprises qui finissent par dénormaliser leurs bases de données relationnelles. Dans le monde relationnel, la dénormalisation viole les formes normales et c'est à éviter. Dans Cassandra, la dénormalisation est tout à fait logique. C'est un avantage supplémentaire car le modèle de données est plus simple, donc plus facile à comprendre.

Dans un SGBDR, on peut facilement changer l'ordre dans lequel les résultats d'une requête SELECT sont retournés, à l'aide de la clause ORDER BY. Mais il n'y a pas d'ordre de tri par défaut, les enregistrements sont retournés dans l'ordre dans lequel ils ont été écrits. Dans Cassandra, le tri est traité différemment : lors de la création de votre ColumnFamily, vous devez définir la stratégie de comparaison qui détermine l'ordre dans lequel les lignes et les colonnes seront triées, mais ce n'est pas définissable dans la requête.

À partir de la version 1.1 d'Apache Cassandra (sortie en septembre 2012), il est possible d'utiliser la clause ORDER BY pour trier le résultat d'une requête sur une colonne donnée (à condition d'utiliser la version 3.0 du langage d'interrogation CQL).

Le site d'Apache Cassandra fournit une explication suffisante pour vous aider à installer, configurer, lancer un cluster Cassandra avec un ou plusieurs nœuds. Voyons de plus près ce que contient la distribution binaire Cassandra :

• bin : contient les exécutables permettant l'exécution de Cassandra et l'interface de ligne de commande cassandra-cli. Ce répertoire contient aussi des scripts pour convertir les fichiers de données (SSTables) en format JSON ou l'inverse ;
• conf : comprend l'ensemble des fichiers nécessaires pour la configuration d'un cluster Cassandra. cassandra.yaml : fichier de configuration principal de Cassandra. Ce fichier contient la configuration du nœud courant, mais aussi celle du cluster et des autres nœuds ;
• interface : le fichier cassandra.thrift est l'interface qui contient la spécification IDL des services offerts pour les API clients Cassandra. Thrift assure l'interopérabilité entre Cassandra et les API implémentant les services décrits dans l'IDL. Ainsi Cassandra peut supporter des clients en Java, C++, PHP, Ruby, Python, C#, via cette interface Thrift ;
• lib : contient toutes les bibliothèques dont Cassandra a besoin pour s'exécuter. On trouve aussi les bibliothèques RPC de Thrift et Avro pour communiquer avec Cassandra.

Cassandra offre depuis les premières versions une interface en ligne de commande très basique : Cassandra-cli. Vous pouvez vous connecter au cluster pour créer ou mettre à jour votre schéma puis définir et récupérer des données de votre base.

Pour se connecter à un cluster Cassandra, c'est simple : il suffit de renseigner le port et l'adresse pour accéder au cluster dont le nom est dans le fichier cassandra.yaml.

cassandra-cli -host 120.0.0.1 -port 9160

#ou

cassandra-cli
[default] connect 127.0.0.1/9160

Pour la création et la manipulation d'un keyspace, d'un column Family…

-- creation d'un keyspace
[default-unknown] create keyspace keyspace1
[default-unknown] use keyspace1

-- création d'un ColumnFamily
[default@keyspace1] CREATE COLUMN FAMILY users
  WITH comparator = UTF8Type
  AND key_validation_class=UTF8Type
  AND column_metadata = [
    {column_name : full_name, validation_class : UTF8Type}
    {column_name : email, validation_class : UTF8Type}
    {column_name : age, validation_class : LongType}
  ];

-- insertion
[default@keyspace1] set User['user1']['first'] = 'Mouhcine';
[default@keyspace1] set User['user1']['last'] = 'MOULOUUU';
[default@keyspace1] set User['user1']['age'] = 27;

-- pour modifier un champ, on utilise la même syntaxe
[default@keyspace1]set User['user1']['last'] = 'MOULOU';

-- Récupération d'une ligne entière par sa clé
[default@keyspace1]get User['user1']

-- ou par autres critères
[default@keyspace1] get User where age > 18;

Avant la version 0.8 de Cassandra, les développeurs Apache n'ont pas trouvé nécessaire d'avoir un langage d'interrogation de données comme SQL. Les clients voulant interroger Cassandra depuis une interface de ligne de commande utilisaient Cassandra-cli.

Avec la sortie de la version 0.8, le langage d'interrogation de données pour Cassandra (CQL) est apparu, dont la syntaxe est basée sur SQL. Par contre, cela ne modifie pas le modèle Cassandra : il n'y a pas de support des jointures, de tri, ni de Group By.

Ci-dessous quelques exemples de requêtes écrites avec CQL :

-- Create a new Keyspace
cqlsh> CREATE KEYSPACE keyspace1
  WITH strategy_class = 'org.apache.cassandra.locator.SimpleStrategy'
  AND strategy_options :replication_factor='1';
cqlsh> USE keyspace1;

-- Create d'un ColumnFamily
cqlsh> CREATE TABLE USERS(
  USER_NAME varchar,
  PASSWORD varchar,
  AGE long,
  PRIMARY KEY (USER_NAME)
);

-- insertion
cqlsh> INSERT INTO USERS  (USER_NAME, PASSWORD) VALUES ('user1', 'p@ssw0rd');

-- Récupération d'une ligne entière par critères
cqlsh> SELECT * FROM users WHERE user_name='user1';

Au cours des premières versions Cassandra, l'interface Thrift était utilisée pour l'implémentation de l'API cliente entière. Malgré ses limitations, Thrift continue d'être supportée dans les nouvelles versions Cassandra.

L'API Thrift offre des fonctionnalités de bas niveau dont bon nombre sont manquantes. Plusieurs implémentations ont vu le jour afin de proposer des API clients Cassandra avec des interfaces de plus haut niveau.

IV-D-1. Hector▲

public class Product {

  private String ref;

  private String name;

  private int quantity;

  private double unitPrice;

  // Constructeur, getters et Setters
}

<dependency>
   <groupId>org.hectorclient</groupId>
   <artifactId>hector-core</artifactId>
   <version>1.1-1</version>
</dependency>

import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;
import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;

// ...

StringSerializer stringSerializer = StringSerializer.get();

DoubleSerializer doubleSerializer = DoubleSerializer.get();

IntegerSerializer integerSerializer = IntegerSerializer.get();

BytesArraySerializer bytesArraySerializer = BytesArraySerializer.get();

Cluster cluster = HFactory.getOrCreateCluster("clusterName", "host");

Keyspace keyspace = HFactory.createKeyspace("keyspaceName", cluster);

ColumnFamilyTemplate<String, String> columnFamilyTemplate = new

ThriftColumnFamilyTemplate<String, String>(keyspace, COLUMN_Family_NAME, stringSerializer, stringSerializer);

Mutator<String> mutator = HFactory.createMutator(keyspace, stringSerializer);
HColumn<String, String> nameColumn = HFactory.createStringColumn("NAME", product.getName());
HColumn<String, Integer> quantityColumn = HFactory.createColumn("QUANTITY", product.getQuantity(), stringSerializer, integerSerializer);

HColumn<String, Double> unitPriceColumn = HFactory.createColumn("UNIT_PRICE", product.getUnitPrice(), stringSerializer, doubleSerializer);

mutator.addInsertion(product.getRef(), COLUMN_Family_NAME,nameColumn)

.addInsertion(product.getRef(), COLUMN_Family_NAME, quantityColumn)

.addInsertion(product.getRef(), COLUMN_Family_NAME, unitPriceColumn);

mutator.execute();

public Product getByRef(String ref){
  checkNotNull(ref, "ref must not be null.");
  logger.info("Getting product its ref where ref= " + ref);

  // les colonnes ont différents datatypes, on utilise donc byte[] pour les valeurs des colonnes avec BytesArraySerializer comme sérialiseur
  RangeSlicesQuery<String, String, byte[]>
  rangeSlicesQuery = HFactory.createRangeSlicesQuery(getKeyspace(), stringSerializer,stringSerializer, bytesArraySerializer);
  rangeSlicesQuery.setColumnFamily(COLUMN_Family_NAME);
  rangeSlicesQuery.setKeys(ref, "")
  rangeSlicesQuery.setRange("", "", false, Integer.MAX_VALUE);
  rangeSlicesQuery.setRowCount(1);
  QueryResult<OrderedRows<String, String, byte[]>> result = range   SlicesQuery.execute();
  OrderedRows<String, String, byte[]> orderedRows = result.get();
  Row<String, String, byte[]>
  row = orderedRows.peekLast();

  if(row.getColumnSlice() == null || row.getColumnSlice().getColum  ns().isEmpty()){
    return null;
  }
  return createProduct(ref, row.getColumnSlice());
}

public Product getByRefCQL(String ref) {
  checkNotNull(ref, "ref must not be null.");
  logger.info("Getting product its ref where ref= " + ref + ", using CQL query.");

  CqlQuery<String, String, byte[]> cqlQuery = new CqlQuery<String, String, byte[]>
(getKeyspace(), stringSerializer, stringSerializer, bytesArraySerializer);

  String query = String.format("Select * from %s where key = '%s' ", COLUMN_Family_NAME, ref);
  cqlQuery.setQuery(query);
  QueryResult<CqlRows<String, String, byte[]>> queryResult = cqlQuery.execute();

  // use 'checkNotNull' (google.guava) to avoid null test with 'if else' clause
  CqlRows<String, String, byte[]> cqlRows = checkNotNull(queryResult.get(), "empty result");
  Row<String, String, byte[]> row = checkNotNull(cqlRows.getByKey(ref), "product with ref :" + ref + "not found");
  return createProduct(ref, row.getColumnSlice());
}

private Product createProduct(String ref, ColumnSlice<String, byte[]>columnSlice) {
  checkNotNull(ref, "ref must not be null.");
  checkArgument((columnSlice!= null &amp;&amp; !columnSlice.getColumns().isEmpty()), "columnsSlice must not be null or empty");
  String name = new String(columnSlice.getColumnByName("NAME").getValue());
  int quantity = ByteBuffer.wrap(columnSlice.getColumnByName("QUANTITY").getValue()).getInt();
  double unitPrice = ByteBuffer.wrap(columnSlice.getColumnByName("UNIT_PRICE").getValue()).getDouble();

  logger.info("initializing new product : [REF=' " + ref +"', NAME=' " + name + "', QUANTITY=" + quantity + ", UNIT_PRICE=" + unitPrice);

  return new Product(ref, name, quantity, unitPrice);
}

public void update(Product product) {
  checkNotNull(product.getRef(), "product.ref must not be null.");
  logger.info("updating row with ref : " + product.getRef());

  ColumnFamilyUpdater<String, String> updater = columnFamilyTemplate.createUpdater(product.getRef());
  updater.setString("NAME", product.getName());
  updater.setInteger("QUANTITY", product.getQuantity());
  updater.setDouble("UNIT_PRICE", product.getUnitPrice());

  try {
    columnFamilyTemplate.update(updater);
  } catch (HectorException e) {
    e.printStackTrace();
    throw new RuntimeException("Unable to update product with key : " +  product.getRef(), e);
  }
}

public void deleteByRef(String ref) {
  checkNotNull(ref, "ref must not be null.");
  logger.info("Deleting product with ref= " + ref);

  columnFamilyTemplate.deleteRow(ref);
}

public void deleteByRefCQL(String ref) {
  checkNotNull(ref, "ref must not be null.");
  logger.info("Deleting product with ref= " + ref + ", using CQL query");

  CqlQuery<String, String, byte[]> cqlQuery = new CqlQuery<String, String, byte[]>(getKeyspace(), stringSerializer, stringSerializer, bytesArraySerializer);
  String query = String.format("Delete from %s where key = '%s' ", COLUMN_Family_NAME, ref);
  cqlQuery.setQuery(query);
  cqlQuery.execute();
}

IV-D-2. Astyanax▲

private static final String COLUMN_FAMILY_NAME ="PRODUCT";
 private static final String NAME_COLUMN = "NAME";
 private static final String QUANTITY_COLUMN = "QUANTITY";
 private static final String UNIT_PRICE_COLUMN = "UNIT_PRICE";
 private static final ColumnFamily<String, String> ProductColumnFamily = getColumnFamily();

 public ProductNoSqlRepository(String clusterName, String host, int port,
 String keyspaceName) {

   AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder()
   .forCluster(clusterName)
   .forKeyspace(keyspaceName)
   .withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
     .setDiscoveryType(NodeDiscoveryType.NONE))
   .withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("samplePool")
   .setPort(port)
   .setMaxConnsPerHost(1)
   .setSeeds(host + " :" + port))
   .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
   .buildKeyspace(ThriftFamilyFactory.getInstance());

   context.start();
   keyspace = context.getEntity();
 }

private static ColumnFamily<String, String> getColumnFamily() {
   return new ColumnFamily<String, String>(COLUMN_FAMILY_NAME, StringSerializer.get(), StringSerializer.get());
 }

 /**
 * @param product
 */
 public void insert(Product product) {
   checkNotNull(product.getRef(), "product.ref must not be null.");

   logger.info("insert new product : " + product.toString());
   try {
     MutationBatch mutationBatch = getKeyspace().prepareMutationBatch();
     mutationBatch.withRow(ProductColumnFamily, product.getRef())
       .putColumn(NAME_COLUMN, product.getName())
       .putColumn(QUANTITY_COLUMN, product.getQuantity())
       .putColumn(UNIT_PRICE_COLUMN, product.getUnitPrice());

    OperationResult<Void> result = mutationBatch.execute();
   } catch (ConnectionException e) {
     logger.error("Unexpected Exception while trying to insert new Product : " + e.getMessage());
   }
 }

/**
 * @param ref
 * @return
 */
 public Product getByRef(String ref) {

   checkNotNull(ref, "ref must not be null.");
   logger.info("Getting product its ref where ref= " + ref);

   Product product = null;
   try {
     OperationResult<ColumnList<String>> result;
     result = getKeyspace().prepareQuery(ProductColumnFamily)
       .getKey(ref)
       .execute();
     ColumnList<String> columns = result.getResult();

     checkArgument(!columns.isEmpty(), "Unable to find product with key : " + ref);
     product = createProduct(ref, columns);

   } catch (ConnectionException e) {
     logger.error("Unexpected Exception while searching for Product with key : " + ref + ".  : -" + e.getMessage());
 }
   return product;
 }

 public void delete(String ref) {

   checkNotNull(ref, "ref must not be null.");
   logger.info("deleting product with ref= " + ref);

   try {
     MutationBatch mutationBatch = getKeyspace().prepareMutationBatch();
     mutationBatch.withRow(ProductColumnFamily, ref).delete();
     mutationBatch.execute();
   } catch (ConnectionException e) {
     logger.error("Unexpected Exception while trying to delete row with product ref : " + ref + ". -" + e);
   }
 }

 /**
 * @param StartFrom
 * @param maxResut
 * @return
 */
 public List<Product> getProducts(String StartFrom, int maxResut){

  List<Product> products = Lists.newArrayList();
  try {
    IndexQuery<String, String> query = getKeyspace()
 .prepareQuery(ProductColumnFamily).searchWithIndex().setStartKey(StartFrom)
 .setRowLimit(maxResut).autoPaginateRows(true);

    Rows<String, String> result = query.execute().getResult();
    for (Iterator<Row<String, String>> iterator = result.iterator(); iterator.hasNext();) {
      Row<String, String> row = iterator.next();
      products.add(createProduct(row.getKey(), row.getColumns()));

    }
   } catch (Exception e) {
     logger.error("" + e);
   }

   return products;
 }

/**
 * @param ref
 * @param columns
 * @return
 */
 private Product createProduct(String ref, ColumnList<String> columns) {
   Product product = new Product(ref);
   product.setName(columns.getColumnByName(NAME_COLUMN).getStringValue());
   product.setQuantity(columns.getColumnByName(QUANTITY_COLUMN).getIntegerValue());
   product.setUnitPrice(columns.getColumnByName(UNIT_PRICE_COLUMN).getDoubleValue());
   return product;
 }

Nous avons souvent l'habitude de créer une BD qui sert à la fois pour les tests unitaires et les tests d'intégration. Chaque poste développeur a donc un cluster Cassandra dédié pour les tests, ou bien les développeurs partagent un même cluster distant (risque de résultat inattendu en cas d'accès concurrent à la même ressource).

Comme les tests SQL en Java utilisent souvent HSQLDB pour mocker la base de données, sur Cassandra il existe un certain nombre de services qui permettent de charger un cluster Cassandra dans la JVM au moment de l'exécution des tests.

IV-E-1. Cassandra-unit▲

<dependency>
<groupId>org.cassandraunit</groupId>
<artifactId>cassandra-unit</artifactId>
<version>1.0.3.1</version>
<scope>test</scope>
</dependency>

{
  "name": "StockKS",
  "replicationFactor": 1,
  "strategy": "org.apache.cassandra.locator.SimpleStrategy",
  "columnFamilies": [
    {
      "name": "PRODUCT",
      "keyType": "UTF8Type",
      "comparatorType": "UTF8Type",
      "columnsMetadata": [
        {
          "name": "NAME",
          "validationClass": "UTF8Type"
        },
        {
          "name": "QUANTITY",
          "validationClass": "IntegerType"
        },
        {
          "name": "UNIT_PRICE",
          "validationClass": "BytesType"
        }
      ]
    }
  ]
}

package fr.soat.it;

import java.io.IOException;

import org.apache.cassandra.config.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.DataLoader;
import org.cassandraunit.dataset.json.ClassPathJsonDataSet;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;

import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;

public abstract class AbstractNoSqlRepositoryIntegrationIT {

 @BeforeClass
 public static void startCassandra() throws TTransportException, IOException, ConfigurationException {
   EmbeddedCassandraServerHelper.startEmbeddedCassandra("cassandra.yaml");
 }

 @Before
 public void setUp() throws IOException,
 TTransportException, ConfigurationException, InterruptedException
 {
   DataLoader dataLoader = new DataLoader("TestCluster",
 "localhost :9160");
   dataLoader.load(new ClassPathJsonDataSet("dataset.json"));

   AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder()
   .forCluster("TestCluster")
   .forKeyspace("test_keyspace")
   .withAstyanaxConfiguration(
   new AstyanaxConfigurationImpl()
   .setDiscoveryType(NodeDiscoveryType.NONE))
   .withConnectionPoolConfiguration(
   new ConnectionPoolConfigurationImpl(
   "testConnectionPool").setPort(9160)
   .setMaxConnsPerHost(1)
   .setSeeds("localhost :9160"))
   .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
 .buildKeyspace(ThriftFamilyFactory.getInstance());

   context.start();
 }

 @After
 public void clearCassandra() {
   EmbeddedCassandraServerHelper.cleanEmbeddedCassandra();
 }

 @AfterClass
 public static void stopCassandra() {
 EmbeddedCassandraServerHelper.stopEmbeddedCassandra();
 }

}

En utilisant la configuration par défaut Cassandra, vous pouvez rapidement rencontrer des problèmes de performance et d'instabilité (consommation de beaucoup de mémoire pour ne faire que des opérations GC par exemple). Il est nécessaire de configurer correctement votre cluster en fonction de vos besoins. Le fichier cassandra.yaml contient tous les attributs nécessaires pour donner plus de performance à votre base. Vous pouvez définir la stratégie et la taille de cache (par clé, par ligne), configurer la taille de la pile Java, l'accès concurrent, etc. Cassandra demande un peu plus de configuration et de surveillance par rapport aux bases de données plus matures.

Bien que Cassandra soit orientée colonne, elle donne plus de performance si on conçoit notre schéma pour stocker plusieurs centaines de colonnes par ligne (exemple : tous les tweets d'un utilisateur stockés dans une seule ligne, avec l'identifiant de l'utilisateur comme clé de la ligne). Mais vous devez être prudent avec cette stratégie d'écriture car elle peut devenir très coûteuse en temps. Une opération d'écriture/lecture peut prendre quelques secondes (et non millisecondes) si votre ligne contient des dizaines de milliers de colonnes, d'où la nécessité de fragmenter les lignes (on peut stocker sur chaque ligne les tweets de l'année en cours, la clé sera donc la concaténation de l'id utilisateur et de l'année).

Nous avons souvent tendance à implémenter des programmes standalone pour les traitements batch afin de supprimer les données obsolètes de notre base. Cassandra offre une alternative qui consiste à définir une valeur TTL (Time-to-live) au moment de la création de la colonne pour préciser la durée pendant laquelle celle-ci sera pérenne. Attention, une fois qu'une valeur TTL est définie, vous ne pouvez plus la modifier.

Cassandra dispose d'une interface de ligne de commande qui repose sur JMX pour vous aider à gérer vos clusters : Nodetool. Il est situé dans le répertoire bin de la distribution binaire Cassandra. Vous pouvez consulter les informations sur votre cluster, faire des snapshots ou afficher les statistiques sur les keyspaces et les columnFamilies.