Como usar o Spark Sql no Cloud Dataproc com metastore no Cloud SQL

Nesta postagem, mostramos como usar o Spark Sql no Cloud Dataproc, sendo que o armazenando dados será feito no Cloud Storage e o metastore em um banco de dados MySQL do Cloud SQL.

O que é Cloud Dataproc?

É a plataforma de big data totalmente gerenciado, para processar grandes quantidades de dados com rapidez, de forma econômica e em grande escala. Usando ferramentas de código aberto como o Apache Spark, o Apache Hive, o Apache Hadoop e o Apache Pig combinadas à escalabilidade dinâmica do Compute Engine e ao armazenamento escalável do Cloud Storage, o Dataproc oferece às equipes analíticas os mecanismos e a elasticidade para executar análises na escala de petabytes por uma fração do custo dos clusters locais tradicional, além de ser facilmente incorporado a outros serviços do Google Cloud Platform (GCP).

O que é Cloud Storage?

Fornece, a nível mundial, um armazenamento de objetos altamente durável que escalona para exabytes de dados. É possível acessar dados instantaneamente de qualquer classe de armazenamento, integrar o armazenamento aos seus aplicativos com uma API exclusiva unificada e otimizar o preço e desempenho de forma fácil.

O que é Apache Hive?

O Apache Hive permite a realizar consultas e análises nos dados que estejam hospedados no sistema de arquivos (HDFS) do Apache Hadoop.

Sendo considerado um Data Warehouse, ele oferece uma linguagem de consulta semelhante ao SQL, que se chama Hive Query Language, na sigla HQL ou HiveQL.

O Hive precisa de um database externo para guardar informações de seus metadados, como por exemplo, as estruturas de suas tabelas ou views. Geralmente os bancos MySQL e PostgreSQL são instalados no cluster para fazer esse armazenamento.

O que é Spark Sql?

O Spark Sql é um modulo do Spark que permite processamento em dados estruturados.

Geralmente é utilizado o mesmo repositório de metadados do Hive.

Existem várias maneiras de interagir com o Spark Sql, por SQL é a mais utilizada, mas é possível conectar por API também.

Independente da forma de interação o mecanismo de execução usado é o mesmo.

O que é Cloud SQL?

É um serviço gerenciado do GCP que permite a criação de um banco de dados relacional MySQL, PostgreSQL ou SQL Server. Na abordagem proposta por esse tutorial, somente os metadados vão estar no Cloud Sql, toda a consulta aos dados será feita pelo Dataproc.

Por que armazenar o metastore no Cloud SQL e os dados no Cloud Storage?

Além de ser uma abordagem que é recomendada pelo próprio GCP, esse desacoplamento oferece diversas possibilidades, entre elas, a "independência" de um cluster, ou seja, com os dados e o metastore guardados externamente, podemos deletar o cluster sem perder essas informações, criar outro com especificações diferentes (mais nós ou memória) para dar continuidade ao trabalho.

É possível também criar um cluster efêmero, e exclui-lo após a conclusão do processo, para redução de custos em processamento de dados não críticos pode-se usar VMs preemptivas.

Passo a Passo - Linha de comando

1. O primeiro passo a fazer é o login em sua conta do GCP, para este tutorial você vai precisar de uma conta que possua um projeto.

2. Vá até a barra superior e clique no ícone do Cloud Shell, igual a imagem abaixo.

3. Agora temos que preparar as variáveis de ambiente que usaremos nos outros passos:

a) Para definir a variável de região usaremos o código abaixo:

export REGION=us-central1

b) Para definir a variável de Zona usaremos o código abaixo:

export ZONE=us-central1-a

4. Nesse passo vamos definir o projeto a ser usado dentro do cloud shell, no campo [PROJECT_ID], insira o id do seu projeto:

gcloud config set project [PROJECT_ID]

5. Nesse passo vamos definir a Zona a ser usada dentro do cloud shell com base na variável que definimos antes:

gcloud config set compute/zone $ZONE

6. Nesse passo vamos definir a Região a ser usada dentro do cloud shell com base na variável que definimos antes:

gcloud config set compute/region $REGION

7. Após isso vamos definir a variável PROJECT com o id do projeto:

export PROJECT=$(gcloud info --format='value(config.project)')

8. Com o código abaixo a api do dataproc será habilitada, caso já esteja pode desconsiderar:

gcloud services enable dataproc.googleapis.com sqladmin.googleapis.com

9. Criaremos o bucket do Cloud Storage que será o repositório de dados do hive:

gsutil mb -l $REGION gs://$PROJECT-warehouse

10. Crie uma instância do Cloud SQL que será usada para armazenar os metadados do hive:

gcloud sql instances create hive-metastore \

--database-version="MYSQL_5_7" \

--activation-policy=ALWAYS \

--gce-zone $ZONE

11. Crie um cluster do dataproc com o comando abaixo:

gcloud dataproc clusters create spark-sql-hive-cluster \

--scopes sql-admin \

--image-version 1.3 \

--initialization-actions gs://dataproc-initialization-actions/cloud-sql-proxy/cloud-sql-proxy.sh \

--properties hive:hive.metastore.warehouse.dir=gs://$PROJECT-warehouse/datasets,spark:hive.metastore.warehouse.dir=gs://$PROJECT-warehouse/datasets,spark:spark.sql.warehouse.dir=gs://$PROJECT-warehouse/datasets \

--metadata "hive-metastore-instance=$PROJECT:$REGION:hive-metastore" \

--region $REGION \

--subnet default \

--master-machine-type n1-standard-1 \

--master-boot-disk-size 500 \

--num-workers 2 \

--worker-machine-type n1-standard-1 \

--worker-boot-disk-size 500 \

--project $PROJECT

Observações:

· Na propriedade “properties” colocamos o que vamos mudar nos arquivos de configuração dos componentes do hadoop, para colocar mais de uma propriedade colocamos o delimitador virgula entre elas.

· A propriedade “hive:hive.metastore.warehouse.dir=gs://$PROJECT-warehouse/datasets” faz referência ao diretório que serão guardadas por padrão os dados provenientes de tabelas gerenciadas em que não foi especificado um PATH, essa mudança é feita no arquivo de configuração do Hive.

· A propriedade “spark:hive.metastore.warehouse.dir=gs://$PROJECT-warehouse/datasets” faz referência ao diretório que serão guardadas por padrão os dados provenientes de tabelas gerenciadas em que não foi especificado um PATH, essa mudança é feita no arquivo de configuração do Spark.

· A propriedade “spark:spark.sql.warehouse.dir=gs://$PROJECT-warehouse/datasets” faz referência ao diretório que serão guardadas por padrão os dados provenientes de tabelas gerenciadas em que não foi especificado um PATH, essa mudança é feita no arquivo de configuração do Spark.

· A propriedade “scopes sql-admin” permite que as vm’s do cluster tenham acesso a API Admin do Cloud SQL.

· A propriedade “image-version 1.3” se refere a versão de imagem de SO do cluster.

· A propriedade “initialization-actions gs://dataproc-initialization-actions/cloud-sql-proxy/cloud-sql-proxy.sh” permite que ao criar o cluster o script de inicialização do Cloud SQL Proxy (que permite fazer acesso às instâncias sem a necessidade de redes autorizadas ou de configuração de SSL de forma segura) execute nas maquinas.

· A propriedade “hive-metastore-instance=$PROJECT:$REGION:hive-metastore” permite que seja estabelecida uma conexão segura com a instância criada do Cloud SQL.

· A propriedade “subnet” faz referência a sub-rede que usaremos no cluster.

· A propriedade “master-machine-type” permite selecionar o tipo de máquina a ser usado pelo master.

· A propriedade “worker-machine-type” permite selecionar o tipo de máquina a ser usado pelos workes.

· A propriedade “project” especifica em qual projeto será criado o cluster.

12. Neste passo vamos gerar um job do Dataproc que vai criar uma tabela externa do hive:

gcloud dataproc jobs submit hive \

--cluster spark-sql-hive-cluster \

--region $REGION \

--execute "

CREATE EXTERNAL TABLE transactions

(SubmissionDate DATE, TransactionAmount DOUBLE, TransactionType STRING)

STORED AS PARQUET

LOCATION 'gs://$PROJECT-warehouse/datasets/transactions';"

Observações:

· Como o metastore será único, qualquer alteração estrutural será refletida tanto no Hive quanto Spark SQL.

· A propriedade “submit” faz referência ao tipo de job que vamos executar, pode-se substituir o valor “hive” por “spark” ou “spark-sql” por exemplo.

· A propriedade “cluster” faz referência ao cluster que executaremos o job.

· A propriedade “region” faz referência a região em que o cluster está.

· A propriedade “execute” contêm o comando a ser executado, nesse caso a instrução de create.

· Caso a instância do Cloud SQL esteja desligada o job vai falhar.

· Podemos ver o resultado do Job através da linha de comando ou no pelo console do GCP.

13. Agora vamos gerar um job do Dataproc com spark sql que vai consultar a tabela externa do hive que criamos no passo anterior:

gcloud dataproc jobs submit spark-sql \

--cluster spark-sql-hive-cluster \

--region $REGION \

--execute "select * from transactions;"

Observação: Também é possível criar os Jobs do Dataproc através do console do GCP.

14. A tabela vai estar vazia pois ainda não tem arquivos na pasta, para que a tabela tenha dados, vamos copiar um arquivo na nossa pasta do Cloud Storage:

gsutil cp gs://hive-solution/part-00000.parquet \

gs://$PROJECT-warehouse/datasets/transactions/part-00000.parquet

15. Execute novamente o job de select para visualizar os dados carregados:

gcloud dataproc jobs submit spark-sql \

--cluster spark-sql-hive-cluster \

--region $REGION \

--execute "select * from transactions limit 10;"

O retorno deve ser semelhante ao abaixo: