Processamento de Dados Massivos/Projeto e implementação de aplicações Big Data/API para processamento estatístico

Descrição da Aplicação

Denominação

API para processamento estatísticos em grandes volumes de dados

Contexto

Extrair informações estatísticas de bases de dados é uma tarefa bastante comum. Bancos relacionais já trazem em suas implementações suporte a diversas funções, tais como soma, média, contabilização, etc.

Em se tratando de grandes volumes de dados, tarefas dessa natureza também são de grande importância. Existem algumas implementações que facilitam essa tarefa, como é o caso do Hive^[1] e Pig^[2], dentre outros. Entretanto, o usuário é obrigado a conhecer detalhes de tecnologias muitas vezes diferentes daquelas que está habituado. Neste contexto, a API aqui proposta visa prover uma série de funções para processamentos estatísticos em grandes volumes de dados, oferecendo ao usuário uma interface bastante simples.

São providas funções corriqueiras como soma, máximo, mínimo, média, desvio padrão e contagem, bem como funções mais elaboradas como particionamento de base, divisão em quantis e criação de tabela de contingência N-dimensional.

Um caso particular de uso é a o processamento da base de dados do SUS para produção de indicadores com vários recortes diferentes, ou ainda sua divisão para validação de algoritmos em um contexto reduzido.

Algoritmo

Os algoritmos aqui descritos operam sobre uma ou mais das seguintes variáveis:

T - base de dados com as transações;

f - condição de filtragem, utilizada para limitar um escopo de trabalho na base T;

i - atributo (ou dimensão) da base T que será alvo do processamento;

i_t - valor do atributo i na transação t;

g - atributos (ou dimensões) que serão usados para geração de agrupamentos. Quando aplicável, cada tupla de valores distintos de g na base T será considerada como critério para agrupamento de registros.

g_i(i=1..N) - cada uma das N tuplas distintas de g na base T. Se g for indefinido, então todo t E T será considerado pertencente a um mesmo grupo g_1.

Máximo

Maior valor assumido pelo atributo i na base T em cada grupo g, limitando as transações t àquelas que atendem ao critério f.

função máximo(T, i, g, f)
    máximo[g_k(k=1..N)] <= VAZIO
    para cada t E T | t atende a f
        se i_t > máximo[g_k | t E g_k] então
            máximo[g_k] <= i_t
        fim se
    fim para
fim função

Mínimo

Menor valor assumido pelo atributo i na base T em cada grupo g, limitando as transações t àquelas que atendem ao critério f.

função mínimo(T, i, g, f)
    mínimo[g_k(k=1..N)] <= VAZIO
    para cada t E T | t atende a f
        se i_t < mínimo[g_k | t E g_k] então
            mínimo[g_k] <= i_t
        fim se
    fim para
fim função

Soma

Soma de todos os valores assumidos pelo atributo i na base T para cada grupo g, limitando as transações t àquelas que atendem ao critério f.

função soma(T, i, g, f)
    soma[g_k(k=1..N)] <= 0
    para cada t E T | t atende a f
        soma[g_k | t E g_k] += i_t
    fim para
fim função

Média

Média dos valores assumidos pelo atributo i na base T em cada grupo g, limitando as transações t àquelas que atendem ao critério f.

função média(T, i, g, f)
    soma[g_k(k=1..N)] <= 0
    elementos[g_k(k=1..N)] <= 0
    para cada t E T | t atende a f
        soma[g_k | t E g_k] += i_t
        elementos[g_k | t E g_k] ++
    fim para
    para cada k (1..N)
        média[g_k] <= soma[g_k] / elementos[g_k]
    fim para
fim função

Contagem

Conta o número de transações t que estão agrupadas em cada grupo g_i, limitando estas transações à apenas aquelas que atendem ao critério f.

função conta(T, g, f)
    conta[g_k(k=1..N)] <= 0
    para cada t E T | t atende a f
        conta[g_k | t E g_k] ++
    fim para
fim função

Desvio padrão

Desvio padrão dos valores assumidos pelo atributo i na base T em cada grupo g, limitando as transações t àquelas que atendem ao critério f.

função desvio(T, i, g, f)
    diferençaQuadrado[g_k(k=1..N)] <= 0
    média(T, i, g, f)
    para cada t E T | t atende a f
        diferençaQuadrado[g_k | t E g_k] += (i_t - média[g_k])^2
        elementos[g_k | t E g_k] ++
    fim para
    para cada k (1..N)
        desvio[g_k] <= raizQuadrada(diferençaQuadrado[g_k] / elementos[g_k])
    fim para
fim função

Particionamento

Particionamento da base em subconjuntos determinados por cada cada grupo g_i, limitando as transações t àquelas que atendem ao critério f. Cada grupo g_i gera uma nova base, composta por todas as transações contida no grupo g_i.

função particiona(T, g, f)
    particao[g_k(k=1..N)] <= VAZIO
    para cada t E T | t atende a f
        particao[g_k | t E g_k] += t
    fim para
fim função

Quantis

Particionamento das transações da base T que atendem ao critério f em subconjuntos, de forma que o intervalo entre o valor mínimo e máximo para o atributo i é dividido em q faixas e cada transação t é alocada no subconjunto equivalente á faixa que se enquadra i_t.

função quantis(T, q, i, f)
    minimo = min(T, q, i, f)
    máximo = max(T, q, i, f)
    f = divide(máximo - mínimo, q)
    quantis[f_k(k=1..q)] <= VAZIO
    para cada t E T | t atende ao critério f
        quantis[f_k | f_(k-1) < i_t <= f_k] += t
    fim para
fim função

Contingência

Gera a tabela de contingência^[3] e as margens para as transações t E T que atendem ao critério f, considerando os atributos definidos em g.

função contingência(T, g, f)
    contingência[g_k(k=1..N)] <= 0
    para cada t E T | t atende a f
        contingência[g_k | t E g_k] ++ 
    fim para
    para cada s | s é subconjunto próprio de g_k(k=1..N)
        margem[s] = 0
        para cada k (1..N)
            se g_k contém s então
                margem[s] += contingência[g_k]
            fim se
        fim para
    fim para
fim função

Exemplo de Funcionamento

Considerando a base de dados T a seguir:

UF	SEXO	ANO	PROCEDIMENTO AMBULATORIAL	VALOR
MG	M	2010	A	10
MG	M	2011	B	2
MG	F	2010	C	4
SP	F	2010	A	10
SP	F	2010	A	10
SP	F	2011	B	2
MT	M	2011	B	2
MT	F	2011	C	4
MS	F	2010	A	10
RS	F	2010	B	2

As saídas para cada comando são:

máximo(T, VALOR, {UF, SEXO}, {PROCEDIMENTO = A})

UF	SEXO	VALOR
MG	M	10.0
MS	F	10.0
SP	F	10.0

mínimo(T, VALOR, {UF, ANO}, NULL)

UF	ANO	VALOR
MG	2010	4.0
MG	2011	2.0
MS	2010	10.0
MT	2011	2.0
RS	2010	2.0
SP	2010	10.0
SP	2011	2.0

soma(T, VALOR, {UF, ANO}, {PROCEDIMENTO EM (A, C)})

UF	ANO	SOMA
MG	2010	14.0
MS	2010	10.0
MT	2011	4.0
SP	2010	20.0

média(T, VALOR, {UF}, NULL)

UF	MÉDIA
MG	5.33
MS	10.0
MT	3.0
RS	2.0
SP	7.33

desvio(T, VALOR, {UF}, NULL)

UF	DESVIO
MG	5.89
MS	0.0
MT	1.41
RS	0.0
SP	6.53

conta(T, {PROCEDIMENTO}, {ANO = 2010})

PROCEDIMENTO	OCORRÊNCIAS
A	4
B	1
C	1

particiona(T, {UF, PROCEDIMENTO}, NULL)

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MT	F	2011	C	4

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MT	M	2011	B	2

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MS	F	2010	A	10

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MG	F	2010	C	4

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MG	M	2010	A	10

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MG	M	2011	B	2

UF	SEXO	ANO	PROCEDIMENTO	VALOR
SP	F	2010	A	10
SP	F	2010	A	10

UF	SEXO	ANO	PROCEDIMENTO	VALOR
SP	F	2011	B	2

UF	SEXO	ANO	PROCEDIMENTO	VALOR
RS	F	2010	B	2

quantis(T, 3, VALOR, NULL)

Nesse exemplo, uma faixa ficou sem representantes

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MG	M	2010	A	10
SP	F	2010	A	10
SP	F	2010	A	10
MS	F	2010	A	10

UF	SEXO	ANO	PROCEDIMENTO	VALOR
MG	M	2011	B	2
MG	F	2010	C	4
SP	F	2011	B	2
MT	M	2011	B	2
MT	F	2011	C	4
RS	F	2010	B	2

contingência(T, {UF, PROCEDIMENTO, ANO}, NULL)

Tabela de contingência

UF	PROCEDIMENTO	ANO	OCORRÊNCIAS
MG	A	2010	1
MG	B	2011	1
MG	C	2010	1
MS	A	2010	1
MT	B	2011	1
MT	C	2011	1
RS	B	2010	1
SP	A	2010	2
SP	B	2011	1

Margens da dimensão PROCEDIMENTO

PROCEDIMENTO	OCORRÊNCIAS
A	4
B	4
C	2

Margens da dimensão UF

UF	OCORRÊNCIAS
MG	3
MS	1
MT	2
RS	1
SP	3

Margens da dimensão ANO

ANO	OCORRÊNCIAS
2010	6
2011	4

Margens das dimensões UF e ANO

UF	ANO	OCORRÊNCIAS
MG	2010	2
MG	2011	1
MS	2010	1
MT	2011	2
RS	2010	1
SP	2010	2
SP	2011	1

Margens das dimensões PROCEDIMENTO e ANO

PROCEDIMENTO	ANO	OCORRÊNCIAS
A	2010	4
B	2010	1
B	2011	3
C	2010	1
C	2011	1

Margens das dimensões UF e PROCEDIMENTO

UF	PROCEDIMENTO	OCORRÊNCIAS
MG	A	1
MG	B	1
MG	C	1
MS	A	1
MT	B	1
MT	C	1
RS	B	1
SP	A	2
SP	B	1

Requisitos

Escalabilidade

As implementações não impõem limite ao volume de dados que são capazes de processar e nem ao número de nós de processamento. Por funcionar sobre a plataforma Hadoop, a escalabilidade da biblioteca depende diretamente da escalabilidade deste último.

Tolerância a falhas

As implementações fornecem o mesmo nível de tolerância a falhas que o sistema de arquivos distribuído HDFS do Hadoop.

Armazenamento

As bases de dados a serem processadas ficam armazenadas no HDFS, que é, por natureza, um sistema de arquivos distribuído construído principalmente para facilitar o acesso aos dados pelos nós de processamento do cluster Hadoop.

Latência

Os algoritmos implementados são pensados para trabalhar em batch, não sendo recomendados para processamento em tempo real, como é comum em operações estatísticas de sistema de banco de dados.

Paralelizações existentes

Tabela de contingência.^[4]

Divisão em quantis.^[5]

Projeto

Oportunidades de paralelização

As funções de agregação simples, tais como máximo, mínimo, soma, contagem e média (soma/contagem) são trivialmente paralelizáveis, uma vez que é possível dividir o conjunto de dados, executar o algoritmo de forma independente sobre cada conjunto, e promover a junção dos resultados parciais. Isto torna o processamento escalável, bastando aumentar o número de nós para comportar um maior volume de dados ou melhorar a performance.

O cálculo do desvio padrão envolve o cálculo da média e posterior cálculo da diferença quadrática de cada elemento para a média. Os dois passos devem ser sequenciais, pois não se pode calcular a diferença antes que a média seja conhecida. No entanto, a média pode ser obtida de forma paralela, bem como a diferença. Neste último caso, a base pode ser distribuída e cada nó é informado do valor da média. Cada nó pode trabalhar de forma independente sobre seu conjunto de dados, produzindo as diferenças quadráticas que, numa fase de redução, servião na composição do cálculo do desvio.

O particionamento também é passível de paralelização, dado que a base de dados pode ser dividida e cada parte pode ser processada de forma independente, separando os registros em cada partição. Um ponto importante a ser levantado aqui é que a quantidade de dados escritos é igual á quantidade de dados lidos, ao contrário das funções de agregação que exigem muito mais leitura que escrita. Haverá aqui uma concorrência na escrita dos registros, já que vários processos podem alimentar uma mesma partição.

Similarmente, a tabela de contingência pode ser construída a partir do processamento independente de blocos da base de dados e posterior junção. As margens podem ser computadas à medida que a tabela é materializada para minimizar comunicação entre os nós, uma vez que os atributos já estarão previamente agrupados.

Para a divisão da base em quantis são necessários dois passos: o primeiro para calcular as faixas de cada quantil e o segundo para separação efetiva dos dados nos respectivos slots. Por serem passos dependentes, devem ser executados de forma sequencial. Entretanto, cada um dos passos pode ser paralelizado. O cálculo das faixas para os quantis passa pela soma e posterior divisão pelo número de faixas. A maior parte da computação, que é a soma, conforme observado anteriormente, pode ser paralelizada. A separação dos registros em quantis pode ser distribuída quebrando-se a base em partes menores e distribuindo aos nós, juntamente com as faixas dos quantis. Cada nó pode processar o que lhe cabe e dar a saída no quantil correto. A escrita deve ser controlada, já que vários processos podem concorrer nesse momento. Também aqui é importante observar que o número de registros lidos e escritos são idênticos, salvo pela utilização de filtros.

Padrões de acesso aos dados

Em todos os casos não há dependência no acesso aos dado visto que são totalmente distribuídos e processados exclusivamente pelos respectivos nós. Há concorrência, em alguns casos, na escrita dos dados. Isso pode ser um gargalo, já que este fenômeno ocorre justamente nas funções que dão saídas a um maior número de registros.

Padrões de comunicação

Nas funções de agregação simples, particionamento e contingência não há necessidade de comunicação entre os nós durante a leitura e processamento, tampouco na fase de agregação.

O particionamento é capaz de escrever no arquivo de saída à medida que os registros são lidos, sendo desnecessárias as comunicações entre os nós (exceto entre os produtores e consumidores).

Para o cálculo do desvio padrão é necessário comunicar o valor da média a todos os nós para que possam, assim, iniciar o processamento.

No caso da divisão em quantis, as faixas claculadas na primeira passada devem ser repassadas a todos os nós, ponto a partir do qual a alocação dos registros poderá ser iniciada.

Linha do tempo integrada

Desenvolvimento

Estratégias de paralelização

Em todas as funções há uma forte caraterística de paralelismo de dados, já que estes são distribuídos. Em nenhuma situação foi observado paralelisme de tarefas sobre os mesmo dados.

Estratégias de armazenamento

O produto das funções é armazenado em disco. Nos casos dos quartis e do desvio padrão, os valores calculados no primeiro passo são deixados em memória primária.

Estratégias de comunicação

Nos casos onde a comunicação é necessária, o agregador dispara todos os leitores do segundo passo já com os valores necessários para os próximos cálculos (média, no desvio padrão e faixas, na divisão em quantis).

Implementação

Plataformas e ferramentas

A solução é implementada sobre as plataformas Hadoop^[6] e Pig^[2], além de fazer uso da biblioteca DataFu^[7].

A linguagem Pig Latin facilita a construção de programas na arquitetura MapReduce de uma forma limpa e intuitiva, sem a necessidade de escrever diretamente as funções map e reduce. Funções como foreach encapsulam mapeamentos, enquanto construções group by mascaram as reduções. No caso de utilização de vários mapeamentos e reduções, a plataforma Pig colabora também na otimização da troca de mensagens. Outro ponto a se considerar aqui é a facilidade de construção e depuração das funções, visto que é possível fazer execução local sem a necessidade de iniciar toda a estrutura Hadoop/HDFS, isto de uma forma bastante transparente.

Soluções com utilização da arquitetura MapReduce diretamente e também da plataforma Hive foram avaliadas. No entanto, a facilidade de desenvolvimento e de manipulação de arquivos oferecidas pelo Pig pesaram a seu favor.

A API é construída na linguagem Java, utilizando a ferramenta Maven^[8] para gerenciar o ciclo de vida do desenvolvimento. Isso possibilita que o módulo seja facilmente incluído como dependência em outros projetos dessa natureza. É possível, ainda, sua utilização em projetos Java que não seguem a estrutura definida pelo Maven e também como programa stand alone.

Integração de plataformas e ferramentas

A API não necessita ser executada diretamente sobre a plataforma Hadoop; a própria plataforma Pig se encarrega de reconhecer e disparar os jobs.

As bases de dados são arquivos CSV carregados no HDFS, utilizando como separador o caracter ponto e vírgula. O próprio sistema de arquivos já faz a distribuição dos blocos entre os diversos nós. Os mapeadores leem dos nós mais próximos e consomem os blocos, comunicando o resultado aos redutores via arquivo no mesmo sistema de arquivos. Nos casos onde são necessárias duas fases de leitura, o final da primeira fase de agregação já dispara os novos mapeamentos, repassando as informações solicitadas ao passo seguinte. O controle da escrita concorrente em arquivos é feita pelo próprio HDFS.

Detalhes de implementação

A função para divisão em quantis utiliza a função de quantis do DataFu para o cálculo das faixas.\

Apesar de várias dessas funções estarem disponíveis ou serem facilmente alcançáveis com a utilização de plataformas como Hive e Pig, a API oculta os detalhes de baixo nível da distribuição, podendo ser utilizada por qualquer desenvolvedor que tenha os conhecimentos básicos de orientação por objeto.

Avaliação

Carga de trabalho

O desempenho da API desenvolvida foi avaliado através da aplicação de suas funcionalidades para analisar uma base de dados de grandes dimensões. Esta base contém centenas de milhões de registros, e a análise estatística destes não é viável utilizando uma abordagem não distribuída.

A fim de avaliar também a escalabilidade da API para bases de dados de tamanhos variados, a base de dados original foi subdividida em várias bases menores. Essa subdivisão permite avaliar o padrão de complexidade das funcionalidades disponibilizadas pela API.

Avaliação experimental

A análise da API foi realizada através de múltiplas execuções de cada uma de suas funcionalidades sobre as bases de dados de diversos tamanhos. Para cada uma das execuções foi analisado o tempo de execução, quantidade de dados lidos e a quantidade de maps e reducers executados.

As funcionalidades testadas estão listadas a seguir, contudo é importante destacar que somente algumas das funcionalidades de agregação foram testadas, pois a maioria possui complexidade semelhante.

Média
Soma
Desvio
Quantis
Contingência
Particionamento

Cada uma das funcionalidades foi testada através de sua execução sobre bases de dados de diversos tamanhos.

Cálculo da média

Tempo de execução
Número de tarefas

Divisão em cinco quantis

Tempo de execução
Número de tarefas

Cálculo do desvio padrão

Tempo de execução
Número de tarefas

Geração da tabela de contingência para três atributos

Tempo de execução
Número de tarefas

Particionamento por três atributos

Tempo de execução
Número de tarefas

Análise de resultados

Os resultados foram obtidos com variação apenas do tamanho da entrada. Outra condição interessante de ser observada é a variação do número de nós de processamento, o que deveria ser feito em trabalhos futuros.

Em todos os casos, é interessante observar o comportamento linear com tendência assintótica dos algoritmos a partir de um determinado volume de registros (4000000, no caso da média). Ademais, o número de mapeadores é sempre significativamente maior que o número de redutores.

Análise Crítica

O comportamento linear assintótico nas funções de agregação sugere que o cluster tende a trabalhar próximo ao seu limite de processamento ao se aumentar o tamanho da entrada. A adição de mais nós, no caso de aumento de dados, deverá subir o limite assintótico, melhorando o desempenho.

O Hadoop assume que as reduções deverão agregar dados, consumindo muito e produzindo pouco. Por esta razão, o número de redutores é significativamente menor que o de mapeadores. Isso é válido para grande parte das funções, mas não é uma assumpção válida para as funções de geração de quantis e particionamento. É possível que o desempenho seja melhorada nesses casos com um melhor balanceamento das tarefas.

Codigo Fonte

O código fonte pode ser encontrado aqui, podendo ser usado e distribuído livremente.

Referências

↑ [1], Hive
↑ ^2,0 ^2,1 [2], Pig
↑ [3], Tabela de contingência
↑ [4], Computing Contingency Statistics in Parallel: Design Trade-Offs and Limiting Cases
↑ [5], Quantiles on Streams
↑ [6], Hadoop
↑ [7], DataFu
↑ [8], Maven

[hive-1] [1], Hive

[pig-2] 2,0 ^2,1 [2], Pig

[contingência-3] [3], Tabela de contingência

[contingencia_paralela-4] [4], Computing Contingency Statistics in Parallel: Design Trade-Offs and Limiting Cases

[quantis_paralelo-5] [5], Quantiles on Streams

[hadoop-6] [6], Hadoop

[datafu-7] [7], DataFu

[maven-8] [8], Maven

[1]

[2]

[3]

[4]

[5]

[6]

[7]

[8]