La bataille des deux Yakozun, ou Cassandra vs HBase. Expérience de l'équipe Sberbank

Ce n'est même pas une blague, il semble que cette image particulière reflète le plus fidèlement l'essence de ces bases de données, et à la fin, il sera clair pourquoi:



Selon DB-Engines Ranking, les deux bases de colonnes NoSQL les plus populaires sont Cassandra (ci-après CS) et HBase (HB).



Par la volonté du destin, notre équipe de gestion du chargement des données à Sberbank travaille en étroite collaboration avec HB depuis longtemps . Pendant ce temps, nous avons assez bien étudié ses forces et ses faiblesses et appris à le cuisiner. Cependant, la présence d'une alternative sous forme de CS tout le temps m'a fait me tourmenter de doutes: avons-nous fait le bon choix? De plus, les résultats de la comparaison effectuée par DataStax indiquent que CS bat facilement HB avec presque un score d'écrasement. D'un autre côté, DataStax est une personne intéressée, et vous ne devriez pas prendre un mot ici. De plus, une assez petite quantité d'informations sur les conditions de test était embarrassante, nous avons donc décidé de découvrir de manière indépendante qui était le roi de BigData NoSql, et les résultats étaient très intéressants.

Cependant, avant de passer aux résultats des tests effectués, il est nécessaire de décrire les aspects essentiels des configurations de l'environnement. Le fait est que CS peut être utilisé en mode de tolérance de perte de données. C'est-à-dire c'est lorsqu'un seul serveur (nœud) est responsable des données d'une certaine clé, et s'il tombe pour une raison quelconque, la valeur de cette clé sera perdue. Pour de nombreuses tâches, ce n'est pas critique, mais pour le secteur bancaire, c'est l'exception plutôt que la règle. Dans notre cas, il est important d'avoir plusieurs copies de données pour un stockage fiable.

Par conséquent, seul le mode CS de triple réplication a été considéré, c'est-à-dire la création de cas a été effectuée avec les paramètres suivants:

CREATE KEYSPACE ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'datacenter1' : 3}; 

De plus, il existe deux façons d'assurer le niveau de cohérence requis. Règle générale:
NW + NR> RF

Cela signifie que le nombre de confirmations des nœuds lors de l'écriture (NW) plus le nombre de confirmations des nœuds lors de la lecture (NR) doit être supérieur au facteur de réplication. Dans notre cas, RF = 3 et donc les options suivantes conviennent:
2 + 2> 3
3 + 1> 3

Puisqu'il est fondamental pour nous de garder les données aussi fiables que possible, un schéma 3 + 1 a été choisi. De plus, HB travaille sur une base similaire, c'est-à-dire une telle comparaison serait plus honnête.

Il convient de noter que DataStax a fait le contraire dans leur étude, ils ont défini RF = 1 pour CS et HB (pour ce dernier en modifiant les paramètres HDFS). C'est un aspect vraiment important, car l'impact sur les performances CS dans ce cas est énorme. Par exemple, l'image ci-dessous montre l'augmentation du temps requis pour charger les données dans CS:



Nous voyons ici ce qui suit, plus les threads concurrents écrivent des données, plus cela prend de temps. C'est naturel, mais il est important que la dégradation des performances pour RF = 3 soit significativement plus élevée. En d'autres termes, si nous écrivons 5 tableaux dans 4 tableaux chacun (20 au total), RF = 3 perd environ 2 fois (150 secondes RF = 3 contre 75 pour RF = 1). Mais si nous augmentons la charge en chargeant les données dans 8 tables dans chacun des 5 flux (40 au total), alors perdre RF = 3 est déjà 2,7 fois (375 secondes contre 138).

C'est peut-être en partie le secret de la réussite des tests de stress DataStax pour CS, car pour HB sur notre stand, le changement du facteur de réplication de 2 à 3 n'a eu aucun effet. C'est-à-dire les disques ne sont pas le goulot d'étranglement pour HB pour notre configuration. Cependant, il existe de nombreux autres pièges, car il convient de noter que notre version de HB a été légèrement corrigée et obscurcie, les environnements sont complètement différents, etc. Il convient également de noter que je ne sais peut-être pas comment préparer correctement CS et qu'il existe des moyens plus efficaces de travailler avec lui et j'espère que dans les commentaires, nous le découvrirons. Mais tout d'abord.

Tous les tests ont été effectués sur un cluster de fer composé de 4 serveurs, chacun dans une configuration:

CPU: Xeon E5-2680 v4 @ 2.40GHz 64 threads.
Disques: 12 morceaux de disque dur SATA
version java: 1.8.0_111


Version CS: 3.11.5


Paramètres GC:


La mémoire jvm.options a été allouée 16 Go (toujours essayé 32 Go, aucune différence n'a été remarquée).

La création de tables a été effectuée par la commande:

 CREATE TABLE ks.t1 (id bigint PRIMARY KEY, title text) WITH compression = {'sstable_compression': 'LZ4Compressor', 'chunk_length_kb': 64}; 

Version HB: 1.2.0-cdh5.14.2 (dans la classe org.apache.hadoop.hbase.regionserver.HRegion nous avons exclu MetricsRegion qui a conduit au GC avec plus de 1000 régions sur RegionServer)


Création de tableaux:
hbase org.apache.hadoop.hbase.util.RegionSplitter ns: t1 UniformSplit -c 64 -f cf
modifier 'ns: t1', {NAME => 'cf', DATA_BLOCK_ENCODING => 'FAST_DIFF', COMPRESSION => 'GZ'}

Il y a un point important - la description DataStax ne dit pas combien de régions ont été utilisées pour créer les tables HB, bien que cela soit critique pour les gros volumes. Par conséquent, pour les tests, le nombre = 64 a été choisi, ce qui permet de stocker jusqu'à 640 Go, soit table de taille moyenne.

Au moment du test, HBase avait 22 000 tables et 67 000 régions (ce serait mortel pour la version 1.2.0, sinon pour le patch mentionné ci-dessus).

Maintenant, pour le code. Puisqu'il n'était pas clair quelles configurations sont plus avantageuses pour une base de données particulière, les tests ont été effectués dans diverses combinaisons. C'est-à-dire dans certains tests, la charge est passée simultanément à 4 tables (les 4 nœuds ont été utilisés pour la connexion). Dans d'autres tests, ils ont travaillé avec 8 tables différentes. Dans certains cas, la taille du lot était de 100, dans d'autres 200 (paramètre de lot - voir code ci-dessous). La taille des données pour la valeur est de 10 octets ou 100 octets (dataSize). Au total, 5 millions d'enregistrements ont été écrits et soustraits à chaque fois dans chaque tableau. Dans le même temps, 5 flux ont été écrits / lus dans chaque table (le numéro de flux est thNum), chacun utilisant sa propre plage de clés (nombre = 1 million):

 if (opType.equals("insert")) { for (Long key = count * thNum; key < count * (thNum + 1); key += 0) { StringBuilder sb = new StringBuilder("BEGIN BATCH "); for (int i = 0; i < batch; i++) { String value = RandomStringUtils.random(dataSize, true, true); sb.append("INSERT INTO ") .append(tableName) .append("(id, title) ") .append("VALUES (") .append(key) .append(", '") .append(value) .append("');"); key++; } sb.append("APPLY BATCH;"); final String query = sb.toString(); session.execute(query); } } else { for (Long key = count * thNum; key < count * (thNum + 1); key += 0) { StringBuilder sb = new StringBuilder("SELECT * FROM ").append(tableName).append(" WHERE id IN ("); for (int i = 0; i < batch; i++) { sb = sb.append(key); if (i+1 < batch) sb.append(","); key++; } sb = sb.append(");"); final String query = sb.toString(); ResultSet rs = session.execute(query); } } 

En conséquence, une fonctionnalité similaire a été fournie pour HB:

 Configuration conf = getConf(); HTable table = new HTable(conf, keyspace + ":" + tableName); table.setAutoFlush(false, false); List<Get> lGet = new ArrayList<>(); List<Put> lPut = new ArrayList<>(); byte[] cf = Bytes.toBytes("cf"); byte[] qf = Bytes.toBytes("value"); if (opType.equals("insert")) { for (Long key = count * thNum; key < count * (thNum + 1); key += 0) { lPut.clear(); for (int i = 0; i < batch; i++) { Put p = new Put(makeHbaseRowKey(key)); String value = RandomStringUtils.random(dataSize, true, true); p.addColumn(cf, qf, value.getBytes()); lPut.add(p); key++; } table.put(lPut); table.flushCommits(); } } else { for (Long key = count * thNum; key < count * (thNum + 1); key += 0) { lGet.clear(); for (int i = 0; i < batch; i++) { Get g = new Get(makeHbaseRowKey(key)); lGet.add(g); key++; } Result[] rs = table.get(lGet); } } 

Étant donné que le client doit veiller à la distribution uniforme des données dans HB, la fonction de salage clé ressemblait à ceci:

 public static byte[] makeHbaseRowKey(long key) { byte[] nonSaltedRowKey = Bytes.toBytes(key); CRC32 crc32 = new CRC32(); crc32.update(nonSaltedRowKey); long crc32Value = crc32.getValue(); byte[] salt = Arrays.copyOfRange(Bytes.toBytes(crc32Value), 5, 7); return ArrayUtils.addAll(salt, nonSaltedRowKey); } 

Maintenant, les résultats les plus intéressants sont:



Identique à un graphique:



L'avantage de HB est tellement incroyable qu'il y a une suspicion d'une sorte de goulot d'étranglement dans les paramètres CS. Cependant, la recherche sur Google et la torsion des paramètres les plus évidents (comme concurrent_writes ou memtable_heap_space_in_mb) n'ont pas donné d'accélération. En même temps, les bûches sont propres, ne jurez rien.

Les données sont réparties uniformément sur les nœuds, les statistiques de tous les nœuds sont approximativement les mêmes.


Une tentative de réduire la taille du lot (jusqu'à l'envoi un par un) n'a pas eu d'effet, elle n'a fait qu'empirer. Il est possible qu'en fait ce soit vraiment la performance maximale pour CS, car les résultats obtenus sur CS sont similaires à ceux obtenus pour DataStax - environ des centaines de milliers d'opérations par seconde. De plus, si vous regardez l'utilisation des ressources, vous verrez que CS utilise beaucoup plus de CPU et de disques:


La figure montre l'utilisation pendant l'exécution de tous les tests consécutifs pour les deux bases de données.

En ce qui concerne les puissants avantages de lecture de HB. On peut voir que pour les deux bases de données, l'utilisation du disque pendant la lecture est extrêmement faible (les tests de lecture sont la dernière partie du cycle de test pour chaque base de données, par exemple, pour CS de 15:20 à 15:40). Dans le cas de HB, la raison est claire - la plupart des données sont bloquées en mémoire, dans memstore, et certaines ont été mises en cache dans blockcache. Quant à CS, il n'est pas très clair comment cela fonctionne, cependant, l'utilisation du disque n'est pas non plus visible, mais juste au cas où une tentative a été faite pour activer le cache row_cache_size_in_mb = 2048 et définir la mise en cache = {'keys': 'ALL', 'row_per_partition': ' 2 000 000 '}, mais cela a encore aggravé la situation.

Il convient également une fois de plus de dire un point significatif sur le nombre de régions de HB. Dans notre cas, la valeur 64 était indiquée. Si vous la réduisez et la rendez égale à par exemple 4, alors lors de la lecture la vitesse diminue de 2 fois. La raison en est que memstore se bouchera plus rapidement et les fichiers seront vidés plus souvent et lors de la lecture, il devra traiter plus de fichiers, ce qui est une opération assez compliquée pour HB. En conditions réelles, cela peut être traité en réfléchissant à la stratégie de pré-plantation et de compactage, en particulier, nous utilisons un utilitaire self-made qui recueille les ordures et compresse les HFiles en permanence en arrière-plan. Il est possible que pour les tests DataStax, généralement 1 région soit allouée par table (ce qui n'est pas correct) et cela clarifierait quelque peu pourquoi HB a tant perdu dans leurs tests de lecture.

Les conclusions préliminaires en sont les suivantes. En supposant qu'aucune erreur grossière n'ait été commise lors des tests, Cassandra est comme un colosse aux pieds d'argile. Plus précisément, alors qu'elle est en équilibre sur une jambe, comme sur la photo au début de l'article, elle montre des résultats relativement bons, mais lorsqu'elle combat dans les mêmes conditions, elle perd carrément. Dans le même temps, compte tenu de la faible utilisation du CPU sur notre matériel, nous avons appris à planter deux RegionServer HB par hôte et ainsi doublé la productivité. C'est-à-dire compte tenu de l'utilisation des ressources, la situation de la CS est encore plus déplorable.

Bien sûr, ces tests sont assez synthétiques et la quantité de données utilisées ici est relativement modeste. Il est possible que lors du passage aux téraoctets, la situation soit différente, mais si pour HB, nous pouvons charger des téraoctets, alors pour CS, cela s'est avéré problématique. Il lançait souvent une OperationTimedOutException même avec ces volumes, bien que les paramètres d'attente de réponse aient déjà été augmentés de plusieurs fois par rapport aux paramètres par défaut.

J'espère que grâce à des efforts conjoints, nous trouverons les goulots d'étranglement CS et si nous parvenons à l'accélérer, j'ajouterai certainement des informations sur les résultats finaux à la fin de l'article.

UPD: Les directives suivantes ont été appliquées lors de la configuration de CS:

disk_optimization_strategy: rotation
MAX_HEAP_SIZE = "32G"
HEAP_NEWSIZE = "3200M"
-Xms32G
-Xmx32G
-XX: + UseG1GC
-XX: G1RSetUpdatingPauseTimePercent = 5
-XX: MaxGCPauseMillis = 500
-XX: InitiatorHeapOccupancyPercent = 70
-XX: ParallelGCThreads = 32
-XX: ConcGCThreads = 8

Quant aux paramètres du système d'exploitation, il s'agit d'une procédure assez longue et compliquée (obtention de root, redémarrage des serveurs, etc.), donc ces recommandations n'ont pas été appliquées. D'un autre côté, les deux bases de données sont dans des conditions égales, donc tout est juste.

Dans la partie code, un connecteur est créé pour tous les threads écrivant dans la table:

 connector = new CassandraConnector(); connector.connect(node, null, CL); session = connector.getSession(); session.getCluster().getConfiguration().getSocketOptions().setConnectTimeoutMillis(120000); KeyspaceRepository sr = new KeyspaceRepository(session); sr.useKeyspace(keyspace); prepared = session.prepare("insert into " + tableName + " (id, title) values (?, ?)"); 

Les données ont été envoyées via une liaison:

 for (Long key = count * thNum; key < count * (thNum + 1); key++) { String value = RandomStringUtils.random(dataSize, true, true); session.execute(prepared.bind(key, value)); } 

Cela n'a pas eu d'impact significatif sur les performances d'enregistrement. Pour la fiabilité, j'ai lancé la charge avec l'outil YCSB, absolument le même résultat. Voici les statistiques pour un thread (sur 4):

2020-01-18 14: 41: 53: 180 315 sec: 10 000 000 d'opérations; 21589,1 opérations courantes / s; [NETTOYAGE: Nombre = 100, Max = 2236415, Min = 1, Moy = 22356,39, 90 = 4, 99 = 24, 99,9 = 2236415, 99,99 = 2236415] [INSÉRER: Nombre = 119551, Max = 174463, Min = 273, Moy = 2582,71, 90 = 3491, 99 = 16767, 99,9 = 99711, 99,99 = 171263]
[GLOBAL], RunTime (ms), 315539
[GLOBAL], Débit (ops / sec), 31691.803548848162
[TOTAL_GCS_PS_Scavenge], nombre, 161
[TOTAL_GC_TIME_PS_Scavenge], temps (ms), 2433
[TOTAL_GC_TIME _% _ PS_Scavenge], Heure (%), 0,7710615803434757
[TOTAL_GCS_PS_MarkSweep], nombre, 0
[TOTAL_GC_TIME_PS_MarkSweep], temps (ms), 0
[TOTAL_GC_TIME _% _ PS_MarkSweep], heure (%), 0,0
[TOTAL_GCs], nombre, 161
[TOTAL_GC_TIME], temps (ms), 2433
[TOTAL_GC_TIME_%], heure (%), 0,7710615803434757
[INSÉRER], Opérations, 10 000 000
[INSÉRER], AverageLatency (us), 3114.2427012
[INSÉRER], MinLatency (us), 269
[INSÉRER], MaxLatency (us), 609279
[INSÉRER], 95thPercentileLatency (us), 5007
[INSÉRER], 99thPercentileLatency (us), 33439
[INSÉRER], Retour = OK, 10000000


Ici, vous pouvez voir que la vitesse d'un flux est d'environ 32 000 enregistrements par seconde, 4 flux ont fonctionné, il en résulte 128 000. Il semble qu'il n'y ait plus rien à presser sur les paramètres actuels du sous-système de disque.

À propos de la lecture plus intéressante. Grâce aux conseils de camarades, il a pu accélérer radicalement. La lecture a été effectuée non pas en 5 flux, mais en 100. Une augmentation à 200 n'a pas produit d'effet. Également ajouté au générateur:
.withLoadBalancingPolicy (nouveau TokenAwarePolicy (DCAwareRoundRobinPolicy.builder (). build ()))

Par conséquent, si auparavant le test a montré 159 644 opérations (5 flux, 4 tables, 100 lots), maintenant:
100 fils, 4 tableaux, lot = 1 (individuellement): 301969 ops
100 fils, 4 tables, lot = 10: 447 608 ops
100 fils, 4 tableaux, lot = 100: 625 655 ops

Étant donné que les résultats sont meilleurs avec les lots, j'ai effectué des tests similaires * avec HB:

* Puisque lorsque vous travaillez dans 400 threads, la fonction RandomStringUtils, qui était utilisée précédemment, charge le CPU à 100%, elle est remplacée par un générateur plus rapide.

Ainsi, une augmentation du nombre de threads lors du chargement des données donne une petite augmentation des performances HB.

Quant à la lecture, voici les résultats de plusieurs options. À la demande de 0x62ash , la commande flush a été exécutée avant la lecture, et plusieurs autres options sont également données à titre de comparaison:
Memstore - lecture à partir de la mémoire, c.-à-d. avant de vider le disque.
HFile + zip - lecture à partir de fichiers compressés par l'algorithme GZ.
HFile + upzip - lecture à partir de fichiers sans compression.

Une caractéristique intéressante est à noter - les petits fichiers (voir le champ «Données», où 10 octets sont écrits) sont traités plus lentement, surtout s'ils sont compressés. Évidemment, cela n'est possible que jusqu'à une certaine taille, évidemment un fichier de 5 Go ne sera pas traité plus rapidement que 10 Mo, mais cela indique clairement que dans tous ces tests, il n'y a toujours pas de champ labouré pour rechercher diverses configurations.

Par intérêt, j'ai corrigé le code YCSB pour travailler avec des lots HB de 100 pièces pour mesurer la latence et plus encore. Ci-dessous est le résultat du travail de 4 copies qui ont écrit sur leurs tables, chacune avec 100 threads. Il s'est avéré ce qui suit:


, CS AverageLatency(us) 3114, HB AverageLatency(us) = 1162 (, 1 = 100 ).

— HBase. , SSD . , , , 4 , 400 , . : — . . ScyllaDB , …