Request is too big

Cassandra
Un moteur in-memory d’écriture
Plan







Caractéristiques principales
Ring
LSM-Tree
Modèle de données
Insertion massive des données
Interrogation des données
Nouveau paradigme: in-memory
Lexique
 Un cluster Cassandra est appelé ring: il fonctionne en mode peer-topeer, chaque nœud du ring pouvant traiter toute demande d’un
client ( absence de relation maître-esclave )
 Un nœud du ring appelé par un client en tant que coordinateur est
capable de lire ou d’écrire des données d’une table ( ou famille de
colonnes ) réparties sur plusieurs noeuds ( architecture de type
shared-nothing )
 Chaque table a ses données répliquées n fois sur les nœuds du
cluster.
 Cassandra optimise l’écriture des données via une table en
mémoire appelée Memtable .
 Les écritures disques se font de manière asynchrone dans une
Sstable ( Sorted String Table )
Caractéristiques
 Solution libre de la fondation Apache développée
initialement par Facebook
 Distribution Datastax ( Community + Enterprise )
 Ecrit en Java
 SGBD orienté colonne => clé-valeur ( valeur =
ensemble de colonnes )
 Système distribué en mode peer-to-peer
Caractéristiques
 Cassandra 2.0
 CQL, système d’interrogation de la base, surcouche
sql => client cqlsh à privilégier au détriment de
cassandra-cli orienté colonne
 Liste des drivers clients: Java, C#, Python
 Pas de locking en cas de mises à jour concurrentes =>
si plusieurs clients modifient les mêmes colonnes de
manière concurrente, seule les modifications les plus
récentes seront conservées.
Une table Cassandra
Caractéristiques
 Atomicité assurée au niveau de la ligne pour une
transaction => insertion et modification de colonnes
pour une ligne traitées comme une seule opération
 Isolation assurée au niveau d’une ligne
 Durabilité assurée via un journal de commit log
 Read & Write consistency
Ring Cassandra
 Système peer-to-peer où chaque nœud est capable de
traiter une demande d’un client ( pas de relation
maître/esclave ).
 Les données des tables sont distribuées de manière
hashée et compressée sur chaque nœud dans des
partitions.
 Chaque partition est répliquée sur des nœuds
différents.
Ring: écriture
Ring: lecture
LSM-tree
LSM-tree
 Structure optimisée pour l’écriture des données, plus
performante qu’une table SQL munie d’index sur de
grands volumes ( GB, TB ).
 Idée principale: écrire en mémoire dans une table de
type clé-valeur, puis écrire sur disque de manière
asynchrone et séquentielle
 Une écriture sur disque est immuable => algorithme
de merge-sort pour fusionner les mêmes tables SST
Ecriture dans une table
Lecture dans une table
Modèle de données
 Ensemble de tables indépendantes les unes des
autres ( pas de jointure en nosql )
 Un seul index, la clé de partition
 Clusterisation possible des tables: clé composite
 Ajout possible d’index secondaires
 Tip: a good rule of a thumb is one column family per
query since you optimize column families for read
performance
Type des données
 Types usuels: int, double, varchar, boolean,
timestamp, blob
 Collections : set, list, map
 Autres types : counter, inet, uuid, timeuuid
Cluster de test
 Cluster à 3 nœuds





I5-3470 ( 4 CPU )
32 GB RAM
4 TB HDD ( 7200 RPM )
Carte réseau à 100 Mb/s
Ubuntu 12.04 LTS
 Commodity hardware
Installation
 Pré-requis:
 Sudo
 JRE Oracle 7
 Accès internet => apt-get
 Installation rapide ( < 1 jour si ports ouverts )
 Documentation:
http://www.datastax.com/documentation/getting_started/
doc/getting_started/gettingStartedDeb_t.html
Insertion massive de données
 Méthode 1 : commande Copy ( cql )
 Import de fichiers csv
 Exemple: copy T from ‘/home/user/file’ with delimiter = ‘|’
 Méthode 2: outil sstableloader
 Générer une SS table à partir d’un fichier csv via un
programme Java à créer
 Utiliser l’outil pour charger la SS table créée dans Cassandra
 Pas d’outil pour insérer des données semi-structurées =>
Création d’un outil en java
SsTableLoad
 SsTableLoad <node_address> <nb_iter> <nb_insert>
<table_name> <min_key>
 Il se connecte à un nœud du ring, lance n itérations sur
une table au format prédéfini.
 Pour chaque itération, il exécute un bulk-insert de m
lignes.
 La première ligne insérée a comme clé min_key, puis on
incrémente de 1 pour chaque nouvelle insertion.
SsTableLoad
 CREATE TABLE test_insert (







string varchar,
nb bigint,
bool boolean,
list list<varchar>,
map map<timestamp,text>,
val blob,
PRIMARY KEY (nb));
 alter table test_insert with gc_grace_seconds = 30;
 Insertion d’un BLOB de 1 MB
Exceptions Java
 Exceptions java rencontrées durant la phase de développement:
 Exception in thread "main"
com.datastax.driver.core.exceptions.NoHostAvailableException:
All host(s) tried for query failed (tried: /192.168.41.26
(com.datastax.driver.core.exceptions.DriverException: Timeout
during read), /192.168.41.71
(com.datastax.driver.core.TransportException: [/192.168.41.71]
Error writing), /192.168.41.86
(com.datastax.driver.core.exceptions.DriverException: Timeout
during read))
 Dans le fichier /etc/cassandra/cassandra.yaml:
 read_request_timeout_in_ms: 5000 => 1 minute
 write_request_timeout_in_ms: 2000 => 24 secondes
Exceptions Java
 Exception in thread "main"
com.datastax.driver.core.exceptions.InvalidQueryExc
eption: Request is too big: length 524366013 exceeds
maximum allowed length 268435456 => nb_insert =
250
 java.lang.OutOfMemoryError: Java heap space =>
changer la taille de la heap size dans le fichier
/etc/cassandra/cassandra-env.sh : 8 GB => 12 GB
Scalabilité
 Un processus charge 64 GB en 14m25s, soit 1 GB en
14s.
 Deux processus chargent 64 GB en 8m47s, soit 1 GB
en 8s.
 Saturation si lancement de 3 processus, un par noeud
Activité réseau
Etude des requêtes
 Objectif: comprendre le fonctionnement interne de
quelques requêtes => tracing on sous cqlsh
 Liste des requêtes étudiées ( CRUD ) :
 Insert
 Update
 Select
 Count(*)
 Scan full, utilisation d’index secondaire, order by
 Delete
Description des tables























CREATE TABLE test_insert_x (
nb bigint,
bool boolean,
"list" list<text>,
"map" map<timestamp, text>,
string text,
val blob,
PRIMARY KEY (nb)
) WITH
bloom_filter_fp_chance=0.010000 AND
caching='ALL' AND
comment='' AND
dclocal_read_repair_chance=0.000000 AND
gc_grace_seconds=30 AND
index_interval=128 AND
read_repair_chance=0.100000 AND
replicate_on_write='true' AND
populate_io_cache_on_flush='false' AND
default_time_to_live=0 AND
speculative_retry='99.0PERCENTILE' AND
memtable_flush_period_in_ms=0 AND
compaction={'class': 'SizeTieredCompactionStrategy'}
AND
compression={'sstable_compression':
'LZ4Compressor'};























CREATE TABLE test_select_x (
nb bigint,
string text,
bool boolean,
"list" list<text>,
"map" map<timestamp, text>,
val blob,
PRIMARY KEY (nb, string)
) WITH
bloom_filter_fp_chance=0.010000 AND
caching='KEYS_ONLY' AND
comment='' AND
dclocal_read_repair_chance=0.000000 AND
gc_grace_seconds=30 AND
index_interval=128 AND
read_repair_chance=0.100000 AND
replicate_on_write='true' AND
populate_io_cache_on_flush='false' AND
default_time_to_live=0 AND
speculative_retry='99.0PERCENTILE' AND
memtable_flush_period_in_ms=0 AND
compaction={'class': 'SizeTieredCompactionStrategy'}
AND
compression={'sstable_compression':
'LZ4Compressor'};
Insert
 insert into test_insert_1 (nb,bool,list,map,string)values (12001,
true, ['azerty', 'qwerty'], { '2014-03-28 12:00' : 't1'}, '12000');
Count(*)
 select count(*) from test_insert_1 limit 20000;
Utilisation de l’index de la clé
primaire
select nb, list, string from test_insert_1 where nb = 1535 ;
Utilisation d’un index secondaire
CREATE INDEX test_insert_1_string_idx ON test_insert_1 (string);
select nb, list, string from test_insert_1 where string =
'VvmEQQwkPEtypCrmBRrKUbhpXXxtfe';
Order by
select nb,string, bool,list,map from test_select_1 where
nb = 1221 order by string
Update
update test_select_1 set bool = true where nb = 1221 and
string = 'LfazkllbGORcyHSwmiZgLVWcmbaWHL' ;
Delete
delete from test_select_1 where nb = 840;
Interrogation des données
 Grammaire du select très limitée, peu d’index, accès
disque en lecture => comment mieux exploiter cette
immense quantité de données collectée ?
 Une solution: version Enterprise de Datastax
 Partie batch ( map-reduce, hive, apache mahout )
 Moteur de recherche full-text: solr ( = elasticsearch )
 Ajout d’une couche in-memory
In-memory
 Changement de paradigme: disque & RAM => RAM &
cache processeur
 Solution 1: coupler Cassandra à un moteur in-memory (
Spark, Shark, MLlib, … )
 Solution 2: coupler Cassandra à une base in-memory en
mode colonne ( Hana de SAP, Vertica de HP, Amazon
RedShift, … ) => cible: BI