Iniciando com Hadoop e Apache Hive: Arquitetura, Configuração e Otimização

Neste artigo busco mostrar como iniciar um ambiente Hadoop com o Apache Hive, como funciona a arquitetura interna do Hive e aplicar alguns recursos de otimização para processamento dos dados com alto desempenho de resposta.

O Apache Hive é uma das ferramentas da Apache Foundation voltado para análise e armazenamento de dados em cenários de big data. Ele utiliza um mecanismo de consultas baseado no modelo MapReduce, no entanto, possui uma camada de abstração do MapReduce para executar consultas em SQL na sua linguagem chamada HiveQL.

Arquitetura do Apache Hive

Em sua arquitetura, temos quatro componentes principais que participam das etapas de operação dentro do Hive:

1. Driver
2. Hive Clients
3. MetaStore
4. Hadoop

Começando pelo Driver, este componente é responsável por analisar a consulta solicitada e fazer a transição para tarefas MapReduce com a ajuda da especificação dos metadados da tabela e do plano de execução gerado posteriormente.

O Hive Clients nada mais é do que a interface para comunicação com o Hive e execução das operações. Neste artigo, estarei utilizando o beeline, no entanto, existem outras como Hue e Hive CLI.

O MetaStore é o catálogo das tabelas presentes no Hive, ele possui informações de banco de dados, tabelas, colunas, partições etc. O MetaStore pode ser armazenado em SGBDs relacionais. Neste artigo estou utilizando o PostgreSQL. Assim que o MetaStore é configurado, todo objeto criado é referenciado nele.

Por fim, o Hadoop são os componentes do ecossistema Hadoop que o Hive utiliza (MapReduce e HDFS) para armazenamento dos dados e execução das consultas. Neste artigo estou utilizando o HDFS do Hadoop para os dados a serem processados.

Iniciando o Hive Server e o Hadoop

Neste artigo, utilizei um Docker Compose com as imagens necessárias para que seja possível fazer esse laboratório do Apache Hive.

version: "3"

services:
  namenode:
    image: bde2020/hadoop-namenode:2.0.0-hadoop2.7.4-java8
    networks:
      - infra-network
    volumes:
      - namenode:/hadoop/dfs/name
    environment:
      - CLUSTER_NAME=test
    env_file:
      - ./hadoop-hive.env
    ports:
      - "50070:50070"

  datanode:
    image: bde2020/hadoop-datanode:2.0.0-hadoop2.7.4-java8
    networks:
      - infra-network
    volumes:
      - datanode:/hadoop/dfs/data
      - ./dados:/hadoop/raw
    env_file:
      - ./hadoop-hive.env
    environment:
      SERVICE_PRECONDITION: "namenode:50070"
    ports:
      - "50075:50075"

  hive-server:
    image: bde2020/hive:2.3.2-postgresql-metastore
    networks:
      - infra-network
    env_file:
      - ./hadoop-hive.env
    environment:
      HIVE_CORE_CONF_javax_jdo_option_ConnectionURL: "jdbc:postgresql://hive-metastore/metastore"
      SERVICE_PRECONDITION: "hive-metastore:9083"
    ports:
      - "10000:10000"

  hive-metastore:
    image: bde2020/hive:2.3.2-postgresql-metastore
    networks:
      - infra-network
    env_file:
      - ./hadoop-hive.env
    command: /opt/hive/bin/hive --service metastore
    environment:
      SERVICE_PRECONDITION: "namenode:50070 datanode:50075 hive-metastore-postgresql:5432"
    ports:
      - "9083:9083"

  hive-metastore-postgresql:
    networks:
      - infra-network
    image: bde2020/hive-metastore-postgresql:2.3.0

  presto-coordinator:
    image: shawnzhu/prestodb:0.181
    networks:
      - infra-network    
    ports:
      - "8080:8080"

networks:
  infra-network:
    driver: bridge

volumes:
  namenode:

  datanode:

Temos nesse compose o datanode e namenode do Hadoop, configurando ambos para as portas padrões 50075 para o datanode e 50070 para o namenode, o hive-server na porta 10000 e o hive-metastore na porta 9083. A imagem utilizada do hive metastore já possui o PostgreSQL configurado. Por fim, temos o PrestoDB que, neste caso, não estarei abordando neste artigo.

Reparem que no serviço do datanode no Docker Compose, estou criando um volume para copiar os arquivos presentes em ./dados para /hadoop/raw. Esses arquivos são o dataset que estou utilizando neste artigo para o laboratório de performance do Apache Hive. O mesmo pode ser encontrado aqui: Kaggle Dataset.

Como esse dataset possui apenas uma única planilha de 78MB, realizei a replicação em 27 planilhas para que as mesmas sejam adicionadas no Hadoop.

Executando o Docker Compose

Após a execução do Docker Compose e todos os serviços estarem iniciados com sucesso, vamos acessar primeiramente o container do datanode Hadoop através do comando:

docker exec -it hive_datanode_1 bash

Acessado com sucesso o container, vamos navegar até a pasta /hadoop/raw para validar se todos nossos arquivos CSV estão contidos.

Validado que os arquivos foram enviados para o container, agora vamos executar o comando para mapear esses arquivos para dentro do HDFS:

hadoop fs -put ./hadoop/raw/ /hadoop/raw/

Após a execução do comando, poderemos ver os arquivos no diretório através do namenode pela URL: localhost:50070.

Conectando no Hive para Criar Tabelas

Agora vamos nos conectar no Hive para criarmos uma tabela em cima dos dados. Para isso devemos acessar o container hive_hive-server_1, acessar o diretório do beeline dentro da pasta do Hive e acessar o Hive através do beeline. Abaixo a sequência de comandos:

1. docker exec -it hive_hive-server_1 bash
2. cd /opt/hive/bin
3. beeline
4. !connect jdbc:hive2://localhost:10000

Será solicitado usuário e senha, pode pressionar enter duas vezes, pois por padrão não há usuário e senha. Após a sequência de comandos, você deverá chegar nesse resultado:

Aqui já estamos “dentro” do Hive e podemos executar nossas operações, como por exemplo: show databases;.

Criei o banco de dados estudo através do comando: CREATE DATABASE ESTUDO; que será o banco onde vou criar as tabelas para os testes de performance com o Hive. Antes de iniciarmos a importação dos arquivos no Hadoop para o Hive, vamos entender um pouco mais sobre as estratégias de otimização que serão aplicadas aqui e os tipos de arquivos que serão testados.

Arquivos do Tipo Apache Parquet

Um arquivo do tipo Parquet é um arquivo em formato binário armazenado de forma colunar. Ele possui um grande desempenho para lidar com dados complexos em massa no ecossistema Hadoop. Ele possui alguns conceitos chave como block size, row group e page. O block size é o tamanho do grupo de linhas que está sendo armazenado em buffer na memória, o row group é o conjunto de linhas que consiste em um bloco de dados para cada coluna (lembrando que Parquet é colunar), já o page é o tamanho da página para compressão, o que facilita a leitura, pois cada página possui um bloco de informações.

Arquivos do Tipo Apache ORC

Os arquivos do tipo Apache ORC são arquivos binários armazenados em formato de dados colunar totalmente otimizados para leitura e gravação no Apache Hive. A grande vantagem do ORC é a forma como os dados são armazenados. Um arquivo ORC possui as chamadas faixas de dados, onde cada faixa possui além de um bloco de dados, informações estatísticas de cada coluna sobre aquele bloco de dados. Isso facilita na busca, pois são utilizadas operações matemáticas como mínimo, máximo, soma de linhas de cada faixa. Além disso, as informações estatísticas são ocasionalmente armazenadas em cache.

Importando os Dados do Hadoop para o Hive

Para nosso laboratório vamos criar um total de quatro tabelas, sendo elas:

• dados_imoveis — tabela principal com os nossos arquivos CSV.
• dados_imoveis_parquet — tabela com nossos dados de origem CSV para Apache Parquet.
• dados_imoveis_orc — tabela com nossos dados de origem CSV para Apache ORC.
• dados_imoveis_orc_partition — tabela com os dados em ORC particionados na sua estrutura de armazenamento no Hadoop por uma coluna específica.

Primeiramente, já conectado ao Hive, vamos importar nossos arquivos CSV para a tabela dados_imoveis.

Criei a tabela dados_imoveis e informei ao Hive que ela está delimitada por vírgula e será armazenada externamente no diretório hdfs://hadoop/raw/dados_imoveis.

Para replicar os dados e atingir o total de 20 milhões de registros, inseri múltiplas vezes os mesmos dados na tabela dados_imoveis.

Após realizar um count na tabela, podemos validar o total de registros.

Em seguida, criei a tabela em Apache Parquet e inseri os registros com base na tabela dados_imoveis.

Por fim, criei a tabela ORC e a tabela ORC particionada pela coluna room, que será atribuída com informações do número de quartos de cada imóvel. Um detalhe importante na sintaxe da criação de uma tabela particionada é que a coluna definida é uma coluna nova e não pode ter o mesmo nome de uma coluna já existente na tabela. Outro detalhe é no momento de inserir os registros, onde a última coluna da inserção deve ser a informação do particionamento.

Particionamento no Hive

No Apache Hive, entre diversos recursos que temos para otimizar nossas operações, existe o particionamento (Partitioned). Basicamente, ele divide a tabela em partes menores com base nos valores da coluna especificada na sua criação e do conteúdo que contém nessa tabela. A vantagem de utilizar esse recurso é a facilidade na busca por fatia dos dados, principalmente quando a busca está sendo feita por uma coluna particionada. A sua estrutura fica da seguinte forma quando criada:

Cada pasta criada representa uma informação de número de quartos, o que garante que apenas os dados relevantes para a consulta sejam lidos, diminuindo não só o tempo de execução, mas também o I/O de disco.

Comparando Performance: CSV x Parquet x ORC

Antes de chegarmos na execução da tabela particionada, vamos avaliar quais tabelas têm melhor desempenho em consultas simples e com agregação.

Inicialmente, realizamos a operação de count sem agregação nas tabelas:

• CSV — 11.438 segundos
• Apache Parquet — 352.529 segundos
• Apache ORC — 5.474 segundos

Neste caso, o Apache Parquet teve um desempenho pior em relação aos demais. Acredito que seja o algoritmo de compactação e codificação do Parquet que, nesse cenário, consumiu mais tempo. Para cada cenário, pode-se obter um desempenho melhor com Parquet ou ORC, mas agora vamos ver como ele se sai com agrupamento e ordenação em relação aos demais.

Para este caso, realizamos o count dos 20 milhões de registros agrupando pela coluna location, ordenando nosso contador de forma decrescente e limitando a 2 registros:

• CSV — 33.868 segundos
• Apache Parquet — 22.945 segundos
• Apache ORC — 12.781 segundos

Já nesse cenário, o Parquet teve um desempenho melhor do que o CSV, o que não aconteceu no cenário anterior. Agora vamos comparar nossa tabela ORC com a tabela ORC particionada pela coluna rooms. Para isso, fizemos uma consulta contando os 20 milhões de registros agrupando pela coluna room, filtrando somente os imóveis que possuem 4 ou 5 quartos e ordenando pelo total de quartos com maior número de imóveis.

Resultado:

• ORC — 7.562 segundos
• ORC particionado — 2.862 segundos

Conclusão

O Apache Hive possui diversas configurações adicionais e outros métodos de otimização que podem ser aplicados para vários cenários. Quem sabe futuramente escreva um artigo somente com o resumo dessas técnicas, afinal cada problema possui uma estratégia diferente de otimização. Abaixo, alguns outros itens que podem ser pesquisados para este mesmo cenário do artigo ou outros cenários:

• Vetorização — Modo de otimizar operações de consulta, filtragem e agregação. Em resumo, a vetorização aumenta o processamento dos blocos de linhas.
• Compressão — Basicamente, ler dados comprimidos. O Hive possui uma inteligência para descompactar na leitura as informações.
• Buckets — Segregar além do particionamento, ou seja, uma flexibilidade de segregar ainda mais os dados.
• Engine Tez — Conforme dito neste artigo, é utilizado o motor de processamento MapReduce. No entanto, é possível utilizar o Tez para acelerar o MapReduce na montagem do plano de consulta.

Espero que você tenha gostado deste artigo e que tenha ficado claro para o seu aprendizado. Qualquer dúvida, pode me contatar no LinkedIn.