Papers
Cassandra: A Decentralized Structured Storage System
Avinash Lakshman, Prashant Malik · ACM SIGOPS Operating Systems Review 44(2) (versao LADIS 2009 lida), 2010
O problema: a falha como regra na escala do Facebook
O paper do Cassandra vem do Facebook, que na época operava uma plataforma social atendendo centenas de milhões de usuários nos horários de pico, a partir de dezenas de milhares de servidores em muitos data centers. Os autores afirmam que lidar com falhas em milhares de componentes é o modo padrão de operação, de modo que o software precisa tratar a falha como a norma, e não como a exceção. O Cassandra foi construído para uma carga concreta, o Inbox Search, que permite aos usuários pesquisar a caixa de entrada do Facebook. Essa funcionalidade exigia uma vazão de escrita muito alta, da ordem de bilhões de escritas por dia, e precisava continuar escalando conforme o número de usuários crescia. Como os usuários são atendidos a partir de data centers geograficamente distribuídos, replicar os dados entre data centers também era fundamental para manter baixa a latência da busca. O paper relata que o Inbox Search foi lançado em junho de 2008 para cerca de 100 milhões de usuários e, na época da escrita, atendia mais de 250 milhões. O abstract descreve o objetivo como um armazenamento distribuído para quantidades muito grandes de dados estruturados em muitos servidores commodity, com serviço altamente disponível e sem ponto único de falha. Ele se parece com um banco de dados em muitas estratégias de projeto, mas não oferece um modelo de dados relacional completo; oferece um modelo mais simples, com controle dinâmico sobre o layout e o formato dos dados. Os autores descrevem o Cassandra como uma síntese de técnicas bem conhecidas para obter escalabilidade e disponibilidade.
Modelo de dados e API
Uma tabela no Cassandra é descrita como um mapa multidimensional distribuído, indexado por uma chave, cujo valor é um objeto altamente estruturado. A chave da linha é uma string sem restrição de tamanho, tipicamente com 16 a 36 bytes. Toda operação sob uma única chave de linha é atômica por réplica, não importa quantas colunas sejam lidas ou escritas. As colunas são agrupadas em famílias de colunas (column families), uma ideia que, segundo os autores, é muito parecida com a do Bigtable. Há dois tipos de famílias de colunas: simples e super, que pode ser imaginada como uma família de colunas dentro de uma família de colunas. As aplicações escolhem como as colunas são ordenadas dentro de uma família, por tempo ou por nome; a ordenação por tempo serve ao Inbox Search, em que os resultados aparecem em ordem cronológica. Uma coluna simples é endereçada como família de colunas e depois coluna, e uma coluna dentro de uma super column family é endereçada como família de colunas, super coluna e depois coluna. O paper observa que as aplicações normalmente usam um cluster Cassandra dedicado só delas e que, embora o sistema aceite várias tabelas, cada implantação tinha apenas uma tabela no seu esquema. A API do cliente é deliberadamente minúscula, com três métodos: insert(table, key, rowMutation), get(table, key, columnName) e delete(table, key, columnName). O argumento do nome da coluna pode apontar para uma coluna específica, uma família de colunas, uma super column family ou uma coluna dentro de uma super coluna.
Particionamento e replicação
O Cassandra particiona os dados com hashing consistente, mas com uma função de hash que preserva a ordem. A faixa de saída é tratada como um anel; cada nó recebe uma posição aleatória nele, e uma chave sofre hash para uma posição e depois percorre o anel no sentido horário até o primeiro nó com posição maior, que passa a ser o coordenador daquela chave. Um nó, portanto, é dono da região entre ele mesmo e o seu predecessor, e a principal vantagem citada pelo paper é que um nó que chega ou sai afeta apenas seus vizinhos imediatos. O paper nomeia duas fraquezas do esquema básico: posições aleatórias dão uma distribuição não uniforme de dados e carga, e o algoritmo ignora diferenças de desempenho entre os nós. Várias posições no anel por nó, ao estilo do Dynamo, são um remédio; o Cassandra, em vez disso, deixa nós pouco carregados se moverem no anel para aliviar os muito carregados, porque os autores constataram que isso tornava o projeto tratável e as escolhas de balanceamento de carga determinísticas. Na replicação, cada item é armazenado em N hosts, em que N é um fator de replicação por instância. O coordenador armazena a chave localmente e a replica em outros N-1 nós. As políticas incluem Rack Unaware, Rack Aware e Datacenter Aware; na Rack Unaware, as outras réplicas são os N-1 sucessores do coordenador no anel. O paper diz que as linhas são replicadas em vários data centers para que a falha de um data center inteiro possa ser tratada sem indisponibilidade. Nas requisições, no modo síncrono as escritas esperam a confirmação de um quorum de réplicas (o paper observa que alguns sistemas de alta vazão usam replicação assíncrona); as leituras vão para a réplica mais próxima ou para todas as réplicas, esperando um quorum, conforme a consistência de que o cliente precisa.
Membership e detecção de falhas
A membership do cluster é baseada no Scuttlebutt, que o paper descreve como um mecanismo de gossip anti-entropia muito eficiente, valorizado pelo baixo uso de CPU e pelo uso eficiente do canal de gossip. O gossip também transporta outros estados de controle do sistema, não apenas a membership. Para a detecção de falhas, os autores usam uma versão modificada do Phi Accrual Failure Detector. Em vez de dar uma resposta sim ou não, o detector produz um nível de suspeita, Phi, numa escala que se ajusta dinamicamente às condições de rede e de carga. Se um nó é suspeito quando o Phi chega a 1, a chance de erro, isto é, de um heartbeat atrasado contradizer depois a decisão, é de cerca de 10 por cento; é de cerca de 1 por cento com Phi igual a 2 e de cerca de 0,1 por cento com 3. Cada nó mantém uma janela deslizante dos intervalos entre chegadas das mensagens de gossip dos outros e calcula o Phi a partir da distribuição resultante. O paper original do accrual sugere uma aproximação gaussiana, mas os autores do Cassandra constataram que uma distribuição exponencial se ajustava melhor, por causa de como o canal de gossip afeta a latência. Eles também afirmam que o detector accrual deles, num cenário de gossip, é, até onde sabem, o primeiro do tipo. Na seção de experiência, outros detectores levaram tempo demais conforme os clusters cresciam: num experimento com 100 nós, detectar um nó falho levou cerca de dois minutos, enquanto o detector accrual com Phi igual a 5 reduziu a média para cerca de 15 segundos.
Persistência local e o caminho de escrita
Em cada máquina, o Cassandra se apoia no sistema de arquivos local. Uma escrita vai primeiro para um commit log, para durabilidade e recuperabilidade, e só depois que isso tem sucesso uma estrutura de dados em memória é atualizada. Os autores dedicam um disco ao commit log porque as escritas nele são sequenciais, o que maximiza a vazão. Quando a estrutura em memória ultrapassa um limiar baseado no tamanho dos dados e no número de objetos, ela é descarregada em disco como um arquivo escrito sequencialmente, junto com um índice pela chave da linha. Com o tempo, muitos desses arquivos se acumulam, e um processo de merge em segundo plano, que o paper diz ser muito parecido com a compactação do Bigtable, os combina. O paper diz que a compactação escolhe arquivos de tamanho semelhante e que uma compactação maior (major compaction) periódica mescla todos os arquivos relacionados em um só, uma operação de uso intenso de I/O. Uma leitura consulta primeiro a estrutura em memória e depois os arquivos em disco, do mais novo para o mais antigo. Para evitar ler arquivos que não têm a chave, cada arquivo de dados tem um bloom filter que resume as suas chaves, também mantido em memória. Os índices de colunas são gerados a cada limite de bloco de 256K, um valor configurável que funcionou bem para eles. Como os arquivos descarregados nunca são modificados, as leituras não precisam de locks, e os autores chamam o servidor de praticamente sem locks (lockless) para leituras e escritas, evitando os problemas de concorrência dos bancos de dados baseados em árvore B. Eles também descrevem um modo de fast sync em que as escritas no commit log são colocadas em buffer, o que arrisca perda de dados se a máquina cair.
Lições operacionais e a implantação do Inbox Search
A seção de experiência do paper abre com a lição de não adicionar nenhuma funcionalidade sem entender como as aplicações a usarão, observando que a maioria dos cenários problemáticos não vem apenas de quedas de nós ou de partições. Antes de lançar o Inbox Search, a equipe indexou 7TB de dados de caixas de entrada de mais de 100 milhões de usuários a partir do MySQL, usando jobs Map/Reduce que atuaram como clientes do Cassandra; canais em segundo plano permitiram a esses jobs agregar o índice reverso de cada usuário, de modo que a instância ficou limitada apenas pela largura de banda da rede. A maioria das aplicações precisava de atomicidade apenas por chave por réplica, mas algumas pediam transações, principalmente para manter índices secundários, e os autores disseram que estavam trabalhando para expor tais operações atômicas. Eles também observaram que até um sistema totalmente descentralizado se beneficia de alguma coordenação, e usaram o Zookeeper para tarefas como eleger um líder que atribui as faixas de réplicas. No Inbox Search, o esquema tinha duas famílias de colunas: uma indexada pelo id do usuário para a busca por termos, em que as palavras são super colunas e os identificadores de mensagens são colunas, e outra para interações, em que os ids de destinatários são super colunas. Quando um usuário clica na barra de busca, uma mensagem assíncrona prepara o cache com o índice daquele usuário. O paper relata cerca de 50+TB num cluster de 150 nós distribuído entre data centers da costa leste e da costa oeste, com latências medidas em milissegundos: para interações, mín 7,69, mediana 15,69, máx 26,13; para busca por termos, mín 7,78, mediana 18,27, máx 44,41. Os trabalhos futuros planejados incluíam compressão, atomicidade entre chaves e índices secundários.
Veja as ideias em ação
Pratique na escola de System Design os desenhos que usam este paper.