A batalha dos dois Yakozun, ou Cassandra vs HBase. Experiência da equipe Sberbank

Isso nem é uma piada, parece que essa imagem em particular reflete com mais precisão a essência desses bancos de dados e, no final, ficará claro o porquê:



De acordo com o DB-Engines Ranking, as duas bases de coluna NoSQL mais populares são Cassandra (daqui em diante CS) e HBase (HB).



Por vontade do destino, nossa equipe de gerenciamento de carregamento de dados no Sberbank trabalha há muito tempo com a HB. Durante esse período, estudamos bem seus pontos fortes e fracos e aprendemos a cozinhá-lo. No entanto, a presença de uma alternativa na forma de CS o tempo todo me fez atormentar-me com dúvidas: fizemos a escolha certa? Além disso, os resultados da comparação realizada pela DataStax disseram que o CS derrota facilmente o HB com quase uma pontuação esmagadora. Por outro lado, o DataStax é uma pessoa interessada e você não deve dar uma palavra aqui. Além disso, uma quantidade bastante pequena de informações sobre as condições de teste foi embaraçosa, por isso decidimos descobrir por nós mesmos quem é o rei do BigData NoSql, e os resultados foram muito interessantes.

No entanto, antes de passar para os resultados dos testes realizados, é necessário descrever os aspectos essenciais das configurações do ambiente. O fato é que o CS pode ser usado no modo de tolerância à perda de dados. I.e. é quando apenas um servidor (nó) é responsável pelos dados de uma determinada chave e, se cair por algum motivo, o valor dessa chave será perdido. Para muitas tarefas, isso não é crítico, mas para o setor bancário, essa é a exceção e não a regra. No nosso caso, é importante ter várias cópias de dados para armazenamento confiável.

Portanto, apenas o modo CS de replicação tripla foi considerado, ou seja, a criação do caso foi realizada com os seguintes parâmetros:

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

Além disso, existem duas maneiras de garantir o nível de consistência necessário. Regra geral:
NW + NR> RF

Isso significa que o número de confirmações dos nós ao escrever (NW) mais o número de confirmações dos nós ao ler (NR) deve ser maior que o fator de replicação. No nosso caso, RF = 3 e, portanto, as seguintes opções são adequadas:
2 + 2> 3
3 + 1> 3

Como é de fundamental importância manter os dados o mais confiável possível, foi escolhido um esquema 3 + 1. Além disso, a HB trabalha em uma base semelhante, ou seja, essa comparação seria mais honesta.

Deve-se notar que o DataStax fez o oposto em sua pesquisa; eles definiram RF = 1 para CS e HB (para o último, alterando as configurações do HDFS). Esse é um aspecto realmente importante, porque o impacto no desempenho do CS neste caso é enorme. Por exemplo, a figura abaixo mostra o aumento no tempo necessário para carregar dados no CS:



Aqui vemos o seguinte: quanto mais threads concorrentes gravam dados, mais tempo leva. Isso é natural, mas é importante que a degradação do desempenho para RF = 3 seja significativamente maior. Em outras palavras, se escrevermos em 4 tabelas em cada um dos 5 fluxos (total de 20), então RF = 3 perde cerca de 2 vezes (150 segundos RF = 3 versus 75 para RF = 1). Mas se aumentarmos a carga carregando dados em 8 tabelas em cada um dos 5 fluxos (total 40), perder RF = 3 já é 2,7 vezes (375 segundos versus 138).

Talvez em parte esse seja o segredo do sucesso do teste de estresse DataStax para CS, porque para a HB em nosso estande, a alteração do fator de replicação de 2 para 3 não teve efeito. I.e. discos não são o gargalo da HB para nossa configuração. No entanto, existem muitas outras armadilhas, porque deve-se notar que nossa versão do HB foi ligeiramente corrigida e obscurecida, os ambientes são completamente diferentes etc. Também vale a pena notar que talvez eu simplesmente não saiba como preparar o CS adequadamente e que haja maneiras mais eficazes de trabalhar com ele, e espero nos comentários que descobriremos. Mas as primeiras coisas primeiro.

Todos os testes foram realizados em um cluster de ferro composto por 4 servidores, cada um em uma configuração:

CPU: Xeon E5-2680 v4 a 2.40GHz 64 threads.
Discos: 12 peças de disco rígido SATA
versão java: 1.8.0_111


Versão CS: 3.11.5


Configurações do GC:


A memória jvm.options recebeu 16 Gb (ainda tentou 32 Gb, nenhuma diferença foi observada).

A criação de tabelas foi realizada pelo comando:

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

Versão do HB: 1.2.0-cdh5.14.2 (na classe org.apache.hadoop.hbase.regionserver.HRegion, excluímos MetricsRegion, que levou ao GC com mais de 1000 regiões no RegionServer)


Criando tabelas:
hbase org.apache.hadoop.hbase.util.RegionSplitter ns: t1 UniformSplit -c 64 -f cf
alterar 'ns: t1', {NOME => 'cf', DATA_BLOCK_ENCODING => 'FAST_DIFF', COMPRESSÃO => 'GZ'}

Há um ponto importante - a descrição do DataStax não diz quantas regiões foram usadas para criar as tabelas HB, embora isso seja crítico para grandes volumes. Portanto, para os testes, foi escolhido o número = 64, o que permite armazenar até 640 GB, ou seja, tabela de tamanho médio.

No momento do teste, o HBase tinha 22 mil tabelas e 67 mil regiões (isso seria fatal para a versão 1.2.0, se não fosse o patch mencionado acima).

Agora para o código. Como não ficou claro quais configurações são mais vantajosas para um banco de dados específico, os testes foram realizados em várias combinações. I.e. em alguns testes, a carga foi simultaneamente para 4 tabelas (todos os 4 nós foram usados ​​para conexão). Em outros testes, eles trabalharam com 8 tabelas diferentes. Em alguns casos, o tamanho do lote era 100, em outros 200 (parâmetro do lote - veja o código abaixo). O tamanho dos dados para o valor é 10 bytes ou 100 bytes (dataSize). No total, 5 milhões de registros foram gravados e subtraídos a cada vez em cada tabela. Ao mesmo tempo, 5 fluxos foram gravados / lidos em cada tabela (o número do fluxo é thNum), cada um dos quais usou seu próprio intervalo de chaves (contagem = 1 milhão):

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

Consequentemente, uma funcionalidade semelhante foi fornecida para a 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); } } 

Como o cliente deve cuidar da distribuição uniforme dos dados no HB, a função de salga principal era assim:

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

Agora, os mais interessantes são os resultados:



O mesmo que um gráfico:



A vantagem do HB é tão surpreendente que há suspeita de algum tipo de gargalo nas configurações de CS. No entanto, pesquisar e torcer os parâmetros mais óbvios (como concurrent_writes ou memtable_heap_space_in_mb) não deu aceleração. Ao mesmo tempo, os logs estão limpos, não juram nada.

Os dados são distribuídos igualmente entre os nós, as estatísticas de todos os nós são aproximadamente as mesmas.


Uma tentativa de reduzir o tamanho do lote (até o envio de um por um) não teve efeito, apenas piorou. É possível que esse seja realmente o desempenho máximo para o CS, pois os resultados obtidos no CS são semelhantes aos obtidos no DataStax - cerca de centenas de milhares de operações por segundo. Além disso, se você observar a utilização dos recursos, verá que o CS usa muito mais CPU e discos:


A figura mostra a utilização durante a execução de todos os testes seguidos nos dois bancos de dados.

Em relação aos poderosos benefícios de leitura da HB. Pode-se observar que, para ambos os bancos de dados, a utilização do disco durante a leitura é extremamente baixa (os testes de leitura são a parte final do ciclo de testes de cada banco de dados, por exemplo, para CS das 15:20 às 15:40). No caso da HB, o motivo é claro - a maioria dos dados fica na memória, no memstore e alguns foram armazenados em cache no blockcache. Quanto ao CS, não está muito claro como ele funciona, no entanto, a utilização do disco também não é visível, mas, no caso, foi feita uma tentativa de ativar o cache row_cache_size_in_mb = 2048 e definir o cache = {'keys': 'ALL', 'lines_per_partition': ' 2.000.000 '}, mas isso piorou ainda mais.

Também vale mais uma vez dizer um ponto significativo sobre o número de regiões no HB. No nosso caso, foi indicado o valor 64. Se você reduzi-lo e torná-lo igual ao exemplo 4, ao ler a velocidade cai 2 vezes. O motivo é que o memstore irá entupir mais rapidamente e os arquivos serão liberados com mais frequência e, ao ler, será necessário processar mais arquivos, o que é uma operação bastante complicada para o HB. Em condições reais, isso pode ser tratado pensando sobre a estratégia de pré-implantação e compactação, em particular, usamos um utilitário criado por você que coleta lixo e comprime HFiles constantemente em segundo plano. É possível que para os testes do DataStax, geralmente 1 região tenha sido alocada por tabela (o que não está correto) e isso esclareceria um pouco por que a HB perdeu tanto em seus testes de leitura.

As conclusões preliminares disso são as seguintes. Supondo que nenhum erro grave tenha sido cometido durante o teste, Cassandra é como um colosso com pés de barro. Mais precisamente, enquanto ela se equilibra sobre uma perna, como na foto no início do artigo, ela mostra resultados relativamente bons, mas quando luta nas mesmas condições, perde totalmente. Ao mesmo tempo, levando em consideração a baixa utilização da CPU em nosso hardware, aprendemos a plantar dois HBs RegionServer por host e, assim, dobramos a produtividade. I.e. levando em consideração a utilização de recursos, a situação do CS é ainda mais deplorável.

Obviamente, esses testes são bastante sintéticos e a quantidade de dados usada aqui é relativamente modesta. É possível que, ao mudar para terabytes, a situação seja diferente, mas se for possível carregar terabytes no HB, então no CS isso se mostrou problemático. Muitas vezes, ele emitia uma OperationTimedOutException mesmo com esses volumes, embora os parâmetros de expectativa de resposta já tivessem aumentado várias vezes em comparação com os padrão.

Espero que, através de esforços conjuntos, encontremos os gargalos do CS e, se conseguirmos acelerá-lo, definitivamente adicionarei informações sobre os resultados finais no final do post.

UPD: As seguintes diretrizes foram aplicadas na configuração do CS:

disk_optimization_strategy: spinning
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

Quanto às configurações do sistema operacional, este é um procedimento bastante longo e complicado (obtenção de raiz, reinicialização de servidores etc.), portanto, essas recomendações não foram aplicadas. Por outro lado, os dois bancos de dados estão em condições iguais, então tudo é justo.

Na parte do código, é criado um conector para todos os threads que estão gravando na tabela:

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

Os dados foram enviados por meio de ligação:

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

Isso não teve um impacto significativo no desempenho da gravação. Para maior confiabilidade, lancei a carga com a ferramenta YCSB, absolutamente o mesmo resultado. Abaixo estão as estatísticas de um segmento (de 4):

2020-01-18 14: 41: 53: 180 315 seg: 10.000.000 operações; 21589.1 ops / s atuais; [LIMPEZA: Contagem = 100, Máx = 2236415, Mín = 1, Média = 22356,39, 90 = 4, 99 = 24, 99,9 = 2236415, 99,99 = 2236415] [INSERIR: Contagem = 119551, Máximo = 174463, Mín = 273, Média = 2582,71, 90 = 3491, 99 = 16767, 99,9 = 99711, 99,99 = 171263]
[GERAL], Tempo de execução (ms), 315539
[GERAL], Taxa de transferência (ops / s), 31691.803548848162
[TOTAL_GCS_PS_Scavenge], contagem, 161
[TOTAL_GC_TIME_PS_Scavenge], tempo (ms), 2433
[TOTAL_GC_TIME _% _ PS_Scavenge], Tempo (%), 0,7710615803434757
[TOTAL_GCS_PS_MarkSweep], contagem, 0
[TOTAL_GC_TIME_PS_MarkSweep], tempo (ms), 0
[TOTAL_GC_TIME _% _ PS_MarkSweep], tempo (%), 0,0
[TOTAL_GCs], contagem, 161
[TOTAL_GC_TIME], tempo (ms), 2433
[TOTAL_GC_TIME_%], tempo (%), 0,7710615803434757
[INSERIR], Operações, 10.000.000
[INSERT], AverageLatency (us), 3114.2427012
[INSERIR], MinLatency (eua), 269
[INSERIR], MaxLatency (us), 609279
[INSERT], 95thPercentileLatency (us), 5007
[INSERT], 99thPercentileLatency (us), 33439
[INSERIR], Retorno = OK, 10000000


Aqui você pode ver que a velocidade de um fluxo é de cerca de 32 mil registros por segundo, 4 fluxos funcionados, resultando em 128 mil. Parece que não há mais nada a se espremer nas configurações atuais do subsistema de disco.

Sobre a leitura mais interessante. Graças ao conselho dos camaradas, ele conseguiu acelerar radicalmente. A leitura foi realizada não em 5 fluxos, mas em 100. Um aumento para 200 não produziu efeito. Também incluído no construtor:
.withLoadBalancingPolicy (novo TokenAwarePolicy (DCAwareRoundRobinPolicy.builder (). build ()))

Como resultado, se anteriormente o teste mostrava 159 644 operações (5 fluxos, 4 tabelas, 100 lotes), agora:
100 threads, 4 tabelas, lote = 1 (individualmente): 301 969 ops
100 threads, 4 tabelas, lote = 10: 447608 ops
100 threads, 4 tabelas, lote = 100: 625 655 ops

Como os resultados são melhores com lotes, executei testes * semelhantes com o HB:

* Como, ao trabalhar em 400 threads, a função RandomStringUtils, usada anteriormente, carregava a CPU em 100%, foi substituída por um gerador mais rápido.

Assim, um aumento no número de threads ao carregar dados fornece um pequeno aumento no desempenho da HB.

Quanto à leitura, aqui estão os resultados de várias opções. A pedido de 0x62ash , o comando flush foi executado antes da leitura e várias outras opções também são fornecidas para comparação:
Memstore - leitura da memória, ou seja, antes de liberar para o disco.
HFile + zip - leitura de arquivos compactados pelo algoritmo GZ.
HFile + upzip - leia arquivos sem compactação.

Um recurso interessante é digno de nota - arquivos pequenos (consulte o campo "Dados", onde são escritos 10 bytes) são processados ​​mais lentamente, especialmente se forem compactados. Obviamente, isso só é possível até um determinado tamanho; obviamente, um arquivo de 5 GB não será processado mais rápido que 10 MB, mas indica claramente que em todos esses testes ainda não há campo para pesquisa de várias configurações.

Por interesse, corrigi o código YCSB para trabalhar com lotes HB de 100 peças para medir a latência e muito mais. Abaixo está o resultado do trabalho de 4 cópias que escreveram em suas tabelas, cada uma com 100 threads. O resultado foi o seguinte:


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

— HBase. , SSD . , , , 4 , 400 , . : — . Você tem que tentar. ScyllaDB , …