Die Schlacht der beiden Yakozun oder Cassandra gegen HBase. Erfahrung im Sberbank-Team

Dies ist nicht einmal ein Scherz, es scheint, dass dieses spezielle Bild die Essenz dieser Datenbanken am genauesten widerspiegelt, und am Ende wird klar sein, warum:



Laut DB-Engines Ranking sind Cassandra (im Folgenden CS) und HBase (HB) die beiden beliebtesten NoSQL-Säulenbasen.



Nach dem Willen des Schicksals arbeitet unser Datenlademanagement-Team bei der Sberbank seit langer Zeit eng mit HB zusammen. In dieser Zeit haben wir seine Stärken und Schwächen sehr gut studiert und gelernt, wie man es kocht. Das ständige Vorhandensein einer Alternative in Form von CS ließ mich jedoch zweifeln: Haben wir die richtige Wahl getroffen? Darüber hinaus haben die Ergebnisse des von DataStax durchgeführten Vergleichs ergeben, dass CS HB leicht mit einem fast vernichtenden Ergebnis besiegt. Auf der anderen Seite ist DataStax eine interessierte Person, und Sie sollten hier kein Wort verlieren. Außerdem war eine kleine Menge an Informationen über die Testbedingungen peinlich, sodass wir beschlossen, selbst herauszufinden, wer der König von BigData NoSql ist, und die Ergebnisse waren sehr interessant.

Bevor Sie jedoch zu den Ergebnissen der durchgeführten Tests übergehen, müssen Sie die wesentlichen Aspekte der Umgebungskonfigurationen beschreiben. Tatsache ist, dass CS im Datenverlust-Toleranzmodus verwendet werden kann. Das heißt In diesem Fall ist nur ein Server (Knoten) für die Daten eines bestimmten Schlüssels verantwortlich. Wenn dieser aus irgendeinem Grund ausfällt, geht der Wert dieses Schlüssels verloren. Für viele Aufgaben ist dies nicht kritisch, für den Bankensektor jedoch eher die Ausnahme als die Regel. In unserem Fall ist es wichtig, mehrere Kopien der Daten für eine zuverlässige Speicherung zu haben.

Daher wurde nur der CS-Modus der dreifachen Replikation in Betracht gezogen, d.h. Die Fallerstellung wurde mit den folgenden Parametern durchgeführt:

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

Darüber hinaus gibt es zwei Möglichkeiten, um das erforderliche Konsistenzniveau sicherzustellen. Allgemeine Regel:
NW + NR> RF

Dies bedeutet, dass die Anzahl der Bestätigungen von den Knoten beim Schreiben (NW) plus die Anzahl der Bestätigungen von den Knoten beim Lesen (NR) größer sein muss als der Replikationsfaktor. In unserem Fall ist RF = 3 und daher sind die folgenden Optionen geeignet:
2 + 2> 3
3 + 1> 3

Da es für uns von grundlegender Bedeutung ist, die Daten so zuverlässig wie möglich zu halten, wurde ein 3 + 1-Schema gewählt. Zusätzlich arbeitet HB auf einer ähnlichen Basis, d.h. Ein solcher Vergleich wäre ehrlicher.

Es sollte beachtet werden, dass DataStax bei seinen Recherchen das Gegenteil getan hat: Sie haben RF = 1 sowohl für CS als auch für HB festgelegt (für letztere durch Ändern der HDFS-Einstellungen). Dies ist ein sehr wichtiger Aspekt, da die Auswirkungen auf die CS-Leistung in diesem Fall enorm sind. Das folgende Bild zeigt zum Beispiel den Anstieg der Zeit, die zum Laden von Daten in CS erforderlich ist:



Hier sehen wir Folgendes: Je mehr konkurrierende Threads Daten schreiben, desto länger dauert es. Dies ist natürlich, aber es ist wichtig, dass die Verschlechterung der Leistung für RF = 3 signifikant höher ist. Mit anderen Worten, wenn wir in 4 Tabellen in jeweils 5 Streams schreiben (insgesamt 20), verliert RF = 3 ungefähr zweimal (150 Sekunden RF = 3 gegenüber 75 für RF = 1). Wenn wir jedoch die Last erhöhen, indem wir Daten in 8 Tabellen in jeweils 5 Streams laden (insgesamt 40), beträgt der Verlust von RF = 3 bereits das 2,7-fache (375 Sekunden gegenüber 138).

Vielleicht ist dies zum Teil das Geheimnis erfolgreicher DataStax-Tests für CS-Lasttests, denn für HB an unserem Stand hatte die Änderung des Replikationsfaktors von 2 auf 3 keine Auswirkung. Das heißt Discs sind in unserer Konfiguration nicht der Flaschenhals für HB. Es gibt jedoch viele andere Fallstricke, da unsere Version von HB leicht gepatcht und verdeckt war, die Umgebungen völlig anders sind usw. Es ist auch erwähnenswert, dass ich vielleicht nicht weiß, wie ich CS richtig vorbereiten soll, und dass es einige effektivere Möglichkeiten gibt, damit zu arbeiten, und ich hoffe, dass wir in den Kommentaren darauf aufmerksam werden. Aber das Wichtigste zuerst.

Alle Tests wurden an einem Eisencluster durchgeführt, der aus 4 Servern mit jeweils einer Konfiguration bestand:

CPU: Xeon E5-2680 v4 @ 2.40GHz 64 Threads.
Festplatten: 12 Stück SATA HDD
Java-Version: 1.8.0_111


CS Version: 3.11.5


GC-Einstellungen:


Dem Speicher jvm.options wurden 16 GB zugewiesen (immer noch 32 GB ausprobiert, es wurde kein Unterschied festgestellt).

Das Erstellen von Tabellen wurde mit dem folgenden Befehl ausgeführt:

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

HB-Version: 1.2.0-cdh5.14.2 (in der Klasse org.apache.hadoop.hbase.regionserver.HRegion haben wir MetricsRegion ausgeschlossen, was zu GC mit mehr als 1000 Regionen auf RegionServer führte)


Tabellen erstellen:
hbase org.apache.hadoop.hbase.util.RegionSplitter ns: t1 UniformSplit -c 64 -f vgl
alter 'ns: t1', {NAME => 'cf', DATA_BLOCK_ENCODING => 'FAST_DIFF', COMPRESSION => 'GZ'}

Es gibt einen wichtigen Punkt: In der DataStax-Beschreibung ist nicht angegeben, wie viele Regionen zum Erstellen der HB-Tabellen verwendet wurden, obwohl dies für große Volumes von entscheidender Bedeutung ist. Daher wurde für die Tests die Zahl = 64 gewählt, wodurch bis zu 640 GB gespeichert werden können, d. H. mittelgroße Tabelle.

Zum Zeitpunkt des Tests verfügte HBase über 22.000 Tabellen und 67.000 Regionen (dies wäre für Version 1.2.0 tödlich, wenn nicht der oben erwähnte Patch).

Nun zum Code. Da nicht klar war, welche Konfigurationen für eine bestimmte Datenbank vorteilhafter sind, wurden die Tests in verschiedenen Kombinationen durchgeführt. Das heißt In einigen Tests wurde die Last gleichzeitig auf 4 Tabellen verteilt (alle 4 Knoten wurden für die Verbindung verwendet). In anderen Tests arbeiteten sie mit 8 verschiedenen Tischen. In einigen Fällen betrug die Chargengröße 100, in anderen 200 (Chargenparameter - siehe Code unten). Die Datengröße für value beträgt 10 Byte oder 100 Byte (dataSize). Insgesamt wurden jedes Mal 5 Millionen Datensätze in jede Tabelle geschrieben und abgezogen. Gleichzeitig wurden 5 Streams in jede Tabelle geschrieben / gelesen (Stream-Nummer ist thNum), von denen jeder seinen eigenen Schlüsselbereich verwendete (Anzahl = 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); } } 

Dementsprechend wurde für HB eine ähnliche Funktionalität bereitgestellt:

 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); } } 

Da der Client für die gleichmäßige Verteilung der Daten in HB sorgen muss, sah die Key-Salting-Funktion folgendermaßen aus:

 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); } 

Am interessantesten sind nun die Ergebnisse:



Das selbe wie ein Graph:



Der Vorteil von HB ist so erstaunlich, dass der Verdacht eines Engpasses in den CS-Einstellungen besteht. Googeln und Torsion der offensichtlichsten Parameter (wie concurrent_writes oder memtable_heap_space_in_mb) führten jedoch nicht zu einer Beschleunigung. Gleichzeitig sind die Protokolle sauber, schwören auf nichts.

Die Daten verteilen sich gleichmäßig auf die Knoten, die Statistiken aller Knoten sind ungefähr gleich.


Der Versuch, die Größe des Stapels zu reduzieren (bis zum Senden nacheinander), hatte keine Auswirkung, es wurde nur noch schlimmer. Es ist möglich, dass dies tatsächlich die maximale Leistung für CS ist, da die mit CS erzielten Ergebnisse mit denen für DataStax vergleichbar sind - etwa Hunderttausende von Vorgängen pro Sekunde. Wenn Sie sich die Ressourcennutzung ansehen, werden Sie außerdem feststellen, dass CS viel mehr CPU und Festplatten benötigt:


Die Abbildung zeigt die Auslastung während des Durchlaufs aller Tests hintereinander für beide Datenbanken.

In Bezug auf die leistungsstarken Lesevorteile von HB. Es ist ersichtlich, dass bei beiden Datenbanken die Festplattenauslastung während des Lesens äußerst gering ist (Lesetests sind der letzte Teil des Testzyklus für jede Datenbank, z. B. für CS von 15:20 bis 15:40 Uhr). Im Fall von HB ist der Grund klar: Die meisten Daten hängen im Speicher, im Memstore, und einige wurden im Blockcache zwischengespeichert. Was CS betrifft, ist nicht ganz klar, wie es funktioniert. Die Festplattenauslastung ist jedoch ebenfalls nicht sichtbar. Es wurde jedoch nur für den Fall versucht, den row_cache_size_in_mb = 2048-Cache einzuschalten und caching = {'keys': 'ALL', 'rows_per_partition': ' 2.000.000 '}, aber das machte es noch ein bisschen schlimmer.

Es lohnt sich auch noch einmal, einen signifikanten Punkt über die Anzahl der Regionen in HB zu sagen. In unserem Fall wurde der Wert 64 angezeigt.Wenn Sie ihn verringern und auf beispielsweise 4 einstellen, sinkt die Geschwindigkeit beim Lesen um das 2-fache. Der Grund ist, dass der Memstore schneller verstopft und die Dateien häufiger gelöscht werden und beim Lesen mehr Dateien verarbeitet werden müssen, was für HB eine recht komplizierte Operation ist. Unter realen Bedingungen kann dies durch Überlegen der Strategie der Voraufteilung und Verdichtung behoben werden. Insbesondere verwenden wir ein selbst erstelltes Dienstprogramm, das den Müll sammelt und HFiles ständig im Hintergrund komprimiert. Es ist möglich, dass für die DataStax-Tests im Allgemeinen 1 Region pro Tabelle zugewiesen wurde (was nicht korrekt ist), und dies würde etwas klarer machen, warum HB bei ihren Lesetests so viel verloren hat.

Die vorläufigen Schlussfolgerungen daraus sind wie folgt. Unter der Annahme, dass beim Testen keine groben Fehler gemacht wurden, ist Cassandra wie ein Koloss mit Lehmfüßen. Genauer gesagt, während sie auf einem Bein balanciert, wie auf dem Bild am Anfang des Artikels, zeigt sie relativ gute Ergebnisse, aber wenn sie unter den gleichen Bedingungen kämpft, verliert sie geradezu. Gleichzeitig haben wir unter Berücksichtigung der geringen CPU-Auslastung unserer Hardware gelernt, zwei RegionServer-HBs pro Host zu installieren und damit die Produktivität zu verdoppeln. Das heißt Unter Berücksichtigung der Ressourcennutzung ist die Situation für CS noch bedauerlicher.

Natürlich sind diese Tests sehr synthetisch und die Datenmenge, die hier verwendet wurde, ist relativ bescheiden. Es ist möglich, dass beim Umschalten auf Terabyte die Situation anders ist, aber wenn wir für HB Terabyte laden können, hat sich dies für CS als problematisch herausgestellt. Auch bei diesen Datenträgern wurde häufig eine OperationTimedOutException ausgelöst, obwohl die Parameter für die Antworterwartung im Vergleich zu den Standardwerten bereits um ein Vielfaches erhöht wurden.

Ich hoffe, dass wir durch gemeinsame Anstrengungen die CS-Engpässe finden und wenn wir es schaffen, sie zu beschleunigen, werde ich definitiv am Ende des Beitrags Informationen über die Endergebnisse hinzufügen.

UPD: Die folgenden Richtlinien wurden beim CS-Setup angewendet:

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

Da dies bei den Betriebssystemeinstellungen ein ziemlich langer und komplizierter Vorgang ist (Root abrufen, Server neu starten usw.), wurden diese Empfehlungen nicht angewendet. Andererseits sind beide Datenbanken unter gleichen Bedingungen, sodass alles fair ist.

Im Codeteil wird ein Connector für alle Threads erstellt, die in die Tabelle schreiben:

 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 (?, ?)"); 

Daten wurden verbindlich übermittelt:

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

Dies hatte keinen wesentlichen Einfluss auf die Aufnahmeleistung. Aus Gründen der Zuverlässigkeit habe ich den Ladevorgang mit dem YCSB-Tool gestartet, absolut das gleiche Ergebnis. Unten ist die Statistik für einen Thread (von 4):

2020-01-18 14: 41: 53: 180 315 Sek .: 10.000.000 Operationen; 21589,1 aktuelle Operationen / Sek .; [CLEANUP: Count = 100, Max = 2236415, Min = 1, Avg = 22356.39, 90 = 4, 99 = 24, 99.9 = 2236415, 99.99 = 2236415] [INSERT: Count = 119551, Max = 174463, Min = 273, Durchschn. = 2582,71, 90 = 3491, 99 = 16767, 99,9 = 99711, 99,99 = 171263]
[OVERALL], RunTime (ms), 315539
[OVERALL], Throughput (ops / sec), 31691.803548848162
[TOTAL_GCS_PS_Scavenge], Count, 161
[TOTAL_GC_TIME_PS_Scavenge], Zeit (ms), 2433
[TOTAL_GC_TIME _% _ PS_Scavenge], Zeit (%), 0,7710615803434757
[TOTAL_GCS_PS_MarkSweep], Count, 0
[TOTAL_GC_TIME_PS_MarkSweep], Zeit (ms), 0
[TOTAL_GC_TIME _% _ PS_MarkSweep], Zeit (%), 0.0
[TOTAL_GCs], Count, 161
[TOTAL_GC_TIME], Zeit (ms), 2433
[TOTAL_GC_TIME_%], Zeit (%), 0,7710615803434757
[EINFÜGEN], Operationen, 10.000.000
[INSERT], AverageLatency (us), 3114.2427012
[INSERT], MinLatency (us), 269
[EINFÜGEN], MaxLatency (us), 609279
[EINFÜGEN], 95thPercentileLatency (us), 5007
[INSERT], 99thPercentileLatency (us), 33439
[INSERT], Return = OK, 10000000


Hier sehen Sie, dass die Geschwindigkeit eines Streams ungefähr 32.000 Datensätze pro Sekunde beträgt, 4 Streams gearbeitet haben und 128.000.Anscheinend gibt es bei den aktuellen Einstellungen des Festplattensubsystems nichts mehr auszudrücken.

Über das Lesen interessanter. Dank der Ratschläge der Genossen konnte er radikal beschleunigen. Das Ablesen wurde nicht in 5, sondern in 100 Strömen durchgeführt. Eine Erhöhung auf 200 hatte keinen Effekt. Ebenfalls zum Builder hinzugefügt:
.withLoadBalancingPolicy (neue TokenAwarePolicy (DCAwareRoundRobinPolicy.builder (). build ())

Als Ergebnis ergab der Test früher 159 644 Operationen (5 Streams, 4 Tabellen, 100 Batch), jetzt:
100 Threads, 4 Tabellen, Batch = 1 (einzeln): 301 969 Ops
100 Threads, 4 Tabellen, Batch = 10: 447 608 Ops
100 Threads, 4 Tabellen, Batch = 100: 625 655 Ops

Da die Ergebnisse bei Chargen besser sind, habe ich ähnliche * Tests mit HB durchgeführt:

* Da bei der Arbeit mit 400 Threads die zuvor verwendete Funktion RandomStringUtils die CPU zu 100% belastete, wurde sie durch einen schnelleren Generator ersetzt.

Daher führt eine Erhöhung der Anzahl der Threads beim Laden von Daten zu einer geringfügigen Erhöhung der HB-Leistung.

Was das Lesen betrifft, sind hier die Ergebnisse mehrerer Optionen. Auf Anforderung von 0x62ash wurde der Befehl flush vor dem Lesen ausgeführt, und es werden auch verschiedene andere Optionen zum Vergleich angegeben:
Memstore - Lesen aus dem Speicher, d.h. vor dem Spülen auf die Festplatte.
HFile + zip - Lesen von Dateien, die mit dem GZ-Algorithmus komprimiert wurden.
HFile + Upzip - Aus Dateien ohne Komprimierung lesen.

Eine interessante Funktion ist bemerkenswert - kleine Dateien (siehe Feld „Daten“, in das 10 Bytes geschrieben werden) werden langsamer verarbeitet, insbesondere wenn sie komprimiert sind. Dies ist natürlich nur bis zu einer bestimmten Größe möglich. Offensichtlich wird eine 5-GB-Datei nicht schneller als 10 MB verarbeitet. Dies zeigt jedoch deutlich, dass in all diesen Tests immer noch kein Feld für die Erforschung verschiedener Konfigurationen vorhanden ist.

Aus Interesse habe ich den YCSB-Code für die Arbeit mit HB-Chargen von 100 Stück korrigiert, um die Latenz und mehr zu messen. Unten ist das Ergebnis der Arbeit von 4 Kopien, die in ihre Tabellen mit jeweils 100 Threads geschrieben haben. Es stellte sich Folgendes heraus:


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

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