Spark Cassandra Connector
bigdata scala / spark / cassandra

Plein de buzzwords !

Le challenge du jour était de commencer à mettre en place toutes ces jolies briques prises séparément pour commencer à se faire une idée précise de comment faire du bigdata en dehors de hadoop, avec :

  • une base de donnée noSQL répartie (Cassandra)
  • un framework de requêtage (Spark) : dispose aussi de librairies SQL-like, d'algorithmes de machine learning, de capacités de temps réel, ...

Comme j'ai des ressources limitées, je fais tout tourner en local, pas de config de prod pour le moment.

Mise en place de Cassandra

Pour cela j'ai suivi un bon tutoriel de Datastax (au passage leur cours sur l'architecture Cassandra est excellent, pour maîtriser les concepts de base)

Au final, démarrage de cassandra en local :

cassandra/bin/cassandra -f

Appel d'un shell interactif de programmation CQL (language de requêtage de type SQL, mais sans jointures, etc...). Pour les gens pressés, ce genre de requêtage avancé, non disponible en général dans les bases noSQL, se fera justement avec Spark ou SparkSQL.

cassandra/bin/cqlsh

Quelques commandes CQL :

DESCRIBE keyspaces;

CREATE KEYSPACE DCO WITH REPLICATION = {'class': 'SimpleStrategy' , 'replication_factor': 1};

use DCO;

CREATE TABLE users (
  user_id int PRIMARY KEY, 
  fname text, 
  lname text
);

INSERT INTO users (user_id,  fname, lname) 
  VALUES (1745, 'john', 'smith');
INSERT INTO users (user_id,  fname, lname) 
  VALUES (1744, 'john', 'doe');
INSERT INTO users (user_id,  fname, lname) 
  VALUES (1746, 'john', 'smith');

Mise en place de Spark

Relativement simple, en partant du site officiel, on arrive à récupérer une version prête à tourner, il suffit de décompresser le .tgz fourni (mais vous pouvez vous le contruire à partir des sources, bien sûr ...)

pour ma part, j'ai pris la version spark-1.3.0-bin-hadoop2.4.tgz

Lancement d'un shell interactif (en scala) :

bin/spark-shell

Lancement d'un shell interactif (en python) :

bin/pyspark

Il est possible d'invoquer les librairies spark en Java, bien sûr.

Connecteur Spark-Cassandra

Jusque là, pas de difficulté particulière pour mettre en place les briques de base. Spark ne possède pas par défaut d'interface native avec des bases de donnée. Un connecteur est développé par Datastax, qui permet de fournir les outils de base pour le big Data, sans focément se reposer sur une infrastructure de type hadoop.

Le principe de mise en oeuvre du connecteur spark-cassandra est simple : on appelle spark-shell en lui fournissant des librairies (.jar) supplémentaires permettant d'accéder à une base de type cassandra. ( http://christopher-batey.blogspot.fr/2015/01/spark-cassandra-basics-connecting-to.html)

Sauf que après plusieurs essais avec diverses versions récupérés pré-compilées, des erreurs lors de l'invocation de la librairie laissaient entendre qu'il manquait toujours une dépendance, ou que la version de telle ou telle brique n'était pas conforme. Après beaucoup d'essais/erreur, (y compris en essayant d'utiliser DSE Datastax Enterprise, qui est censé intégrer tout cela, mais que je n'ai pas pris le temps de faire fonctionner correctement), je laisse tomber cette piste, et reviens aux fondamentaux des briques opensources :

La recompilation !

La page du projet qui m'a permis d'avancer :

https://github.com/datastax/spark-cassandra-connector/blob/master/doc/13_spark_shell.md

En gros, récupération du source, téléchargement des dépendances (merci les proxys d'entreprise ...), installation des outils de compilation java que je n'avais pas encore (merci Ubuntu !), et au bout d'un moment, le message de confirmation de la compilation, et les beaux .jar tous neufs disponibles sous spark-cassandra-connector/target/scala-2.10

Lancement du spark-shell :

bin/spark-shell --jars /home/daniel/spark-cassandra-connector-master/spark-cassandra-connector/target/scala-2.10/spark-cassandra-connector-assembly-1.3.0-SNAPSHOT.jar,/home/daniel/spark-cassandra-connector-master/spark-cassandra-connector-java/target/scala-2.10/spark-cassandra-connector-java-assembly-1.3.0-SNAPSHOT.jar --conf spark.cassandra.connection.host=127.0.0.1

une fois le prompt scala disponible :

import com.datastax.spark.connector._ //Imports basic rdd functions
import com.datastax.spark.connector.cql._ //(Optional) Imports java driver helper functions
val rdd=sc.cassandraTable("dco", "table1")
rdd.toArray.foreach(println)

On accède aux données d'une table précédemment créé sous cassandra / CQL !

Il est possible d'invoquer directement des commandes CQL depuis spark, bien sûr :

val c = CassandraConnector(sc.getConf)
c.withSessionDo ( session => session.execute("CREATE KEYSPACE test WITH replication={'class':'SimpleStrategy', 'replication_factor':1}"))
c.withSessionDo ( session => session.execute("CREATE TABLE test.fun (k int PRIMARY KEY, v int)"))
sc.parallelize(1 to 100).map( x => (x,x)).saveToCassandra("test","fun")
sc.cassandraTable("test","fun").take(3)

Voilà pour une première mise en place, on va pouvoir faire des choses plus complexes ... (à suivre)

- Tintouli