1. Introdução: Google Dataproc
O Dataproc é um serviço totalmente gerenciado e altamente escalonável para executar o Apache Spark, o Apache Flink, o Presto e muitas outras ferramentas e frameworks de código aberto. Use o Dataproc para modernização do data lake, ETL / ELT e ciência de dados segura em escala planetária. O Dataproc também é totalmente integrado a vários serviços do Google Cloud, incluindo BigQuery, Cloud Storage, Vertex AI e Dataplex.
O Dataproc está disponível em três versões:
- Com o Dataproc sem servidor, é possível executar jobs do PySpark sem precisar configurar infraestrutura e escalonamento automático. O Dataproc sem servidor oferece suporte a sessões e cargas de trabalho em lote do PySpark (notebooks).
- O Dataproc no Google Compute Engine permite gerenciar um cluster Hadoop YARN para cargas de trabalho do Spark baseadas em YARN, além de ferramentas de código aberto como Flink e Presto. É possível personalizar seus clusters baseados em nuvem com o escalonamento vertical ou horizontal que quiser, incluindo o escalonamento automático.
- O Dataproc no Google Kubernetes Engine permite configurar clusters virtuais do Dataproc na infraestrutura do GKE para enviar jobs do Spark, PySpark, SparkR ou Spark SQL.
2. Criar um cluster do Dataproc em uma VPC do Google Cloud
Nesta etapa, você vai criar um cluster do Dataproc no Google Cloud usando o console do Google Cloud.
Como primeira etapa, ative a API de serviço do Dataproc no console do Google Cloud. Depois de ativá-lo, pesquise "Dataproc" na barra de pesquisa e clique em Criar cluster.
Selecione Cluster no Compute Engine para usar VMs do Google Compute Engine(GCE) como a infraestrutura subjacente para executar clusters do Dataproc.
Agora você está na página de criação de cluster.
Nesta página:
- Dê um nome exclusivo ao cluster.
- Selecione a região específica. Você também pode selecionar uma zona, mas o Dataproc oferece a capacidade de escolher uma automaticamente. Para este codelab, selecione "us-central1" e "us-central1-c".
- Selecione o tipo de cluster "Padrão". Isso garante que haja um nó principal.
- Na guia Configurar nós, confirme se o número de workers criados será dois.
- Na seção Personalizar cluster, marque a caixa ao lado de Ativar gateway de componentes. Isso permite o acesso a interfaces da Web no cluster, incluindo a interface do Spark, o gerenciador de nós do Yarn e os notebooks do Jupyter.
- Em Componentes opcionais, selecione Notebook Jupyter. Isso configura o cluster com um servidor de notebooks do Jupyter.
- Deixe tudo no estado em que se encontra e clique em Criar cluster.
Isso vai ativar um cluster do Dataproc.
3. Iniciar o cluster e usar SSH para acessar
Quando o status do cluster mudar para Em execução, clique no nome do cluster no console do Dataproc.
Clique na guia Instância de VM para conferir o nó mestre e os dois nós de trabalho do cluster.
Clique em SSH ao lado do nó mestre para fazer login nele.
Execute comandos hdfs para ver a estrutura de diretórios.
hadoop_commands_example
sudo hadoop fs -ls /
sudo hadoop version
sudo hadoop fs -mkdir /test51
sudo hadoop fs -ls /
4. Interfaces da Web e gateways de componentes
No console de cluster do Dataproc, clique no nome do cluster e depois na guia INTERFACES DA WEB.
Isso mostra as interfaces da Web disponíveis, incluindo o Jupyter. Clique em Jupyter para abrir um notebook do Jupyter. Você pode usar isso para criar notebooks em PySpark armazenados no GCS. para armazenar seu notebook no Google Cloud Storage e abrir um notebook PySpark para usar neste codelab.
5. Monitorar e observar jobs do Spark
Com o cluster do Dataproc em execução, crie um job em lote do PySpark e envie-o para o cluster do Dataproc.
Crie um bucket do Google Cloud Storage (GCS) para armazenar o script do PySpark. Crie o bucket na mesma região do cluster do Dataproc.
Agora que o bucket do GCS foi criado, copie o arquivo a seguir para ele.
https://raw.githubusercontent.com/diptimanr/spark-on-gce/main/test-spark-1.py
Esse script cria um DataFrame do Spark de amostra e o grava como uma tabela do Hive.
hive_job.py
from pyspark.sql import SparkSession
from datetime import datetime, date
from pyspark.sql import Row
spark = SparkSession.builder.master("local").enableHiveSupport().getOrCreate()
df = spark.createDataFrame([ (1, 2., 'string1', date(2000, 1, 1), datetime(2000, 1, 1, 12, 0)),
(2, 3., 'string2', date(2000, 2, 1), datetime(2000, 1, 2, 12, 0)), (3, 4., 'string3', date(2000, 3, 1), datetime(2000, 1, 3, 12, 0))
], schema='a long, b double, c string, d date, e timestamp')
print("..... Writing data .....")
df.write.mode("overwrite").saveAsTable("test_table_1")
print("..... Complete .....")
Envie esse script como um job em lote do Spark no Dataproc. Clique em Jobs no menu de navegação à esquerda e em Enviar job.
Forneça um código da tarefa e uma região. Selecione o cluster e informe o local do GCS do script do Spark que você copiou. Esse job será executado como um job em lote do Spark no Dataproc.
Em Propriedades, adicione a chave spark.submit.deployMode e o valor client para garantir que o driver seja executado no nó principal do Dataproc e não nos nós de trabalho. Clique em Enviar para enviar o job em lote ao Dataproc.
O script do Spark vai criar um DataFrame e gravar em uma tabela do Hive test_table_1.
Depois que o job for executado com sucesso, você poderá ver as instruções de impressão do console na guia Monitoring.
Agora que a tabela do Hive foi criada, envie outro job de consulta do Hive para selecionar o conteúdo da tabela e mostrar no console.
Crie outro job com as seguintes propriedades:
Observe que o Tipo de job está definido como Hive e o tipo de origem da consulta é Texto da consulta, o que significa que vamos escrever toda a instrução HiveQL na caixa de texto Texto da consulta.
Envie o job, mantendo o restante dos parâmetros como padrão.
Observe como o HiveQL seleciona todos os registros e os mostra no console.
6. Escalonamento automático
O escalonamento automático é a tarefa de estimar o número "certo" de nós de trabalho do cluster para uma carga de trabalho.
A API AutoscalingPolicies do Dataproc fornece um mecanismo para automatizar o gerenciamento de recursos do cluster e permite o escalonamento automático da VM de worker do cluster. Uma política de escalonamento automático é uma configuração reutilizável que descreve como os workers do cluster que usam a política de escalonamento automático precisam ser escalonados. Ele define limites, frequência e agressividade de escalonamento para fornecer controle detalhado dos recursos do cluster em todo o respectivo ciclo de vida.
As políticas de escalonamento automático do Dataproc são gravadas usando arquivos YAML, que são transmitidos no comando da CLI para criar o cluster ou selecionados em um bucket do GCS quando um cluster é criado no console do Cloud.
Confira um exemplo de uma política de escalonamento automático do Dataproc :
policy.yaml
workerConfig:
minInstances: 10
maxInstances: 10
secondaryWorkerConfig:
maxInstances: 50
basicAlgorithm:
cooldownPeriod: 4m
yarnConfig:
scaleUpFactor: 0.05
scaleDownFactor: 1.0
gracefulDecommissionTimeout: 1h
7. Configurar componentes opcionais do Dataproc
Isso vai ativar um cluster do Dataproc.
Quando você cria um cluster do Dataproc, os componentes padrão do ecossistema Apache Hadoop são instalados automaticamente (consulte a Lista de versões do Dataproc). Ao criar um cluster, é possível instalar componentes adicionais, chamados de Componentes opcionais.
Ao criar o cluster do Dataproc no console, ativamos os componentes opcionais e selecionamos Notebook do Jupyter como o componente opcional.
8. Limpar recursos
Para limpar o cluster, clique em Parar depois de selecionar o cluster no console do Dataproc. Quando o cluster for interrompido, clique em Excluir para excluí-lo.
Depois que o cluster do Dataproc for excluído, exclua os buckets do GCS em que o código foi copiado.
Para liberar memória dos recursos e interromper qualquer faturamento indesejado, primeiro pare e depois exclua o cluster do Dataproc.
Antes de parar e excluir o cluster, verifique se todos os dados gravados no armazenamento HDFS foram copiados para o GCS para armazenamento durável.
Para interromper o cluster, clique em Parar.
Quando o cluster for interrompido, clique em Excluir para excluí-lo.
Na caixa de diálogo de confirmação, clique em Excluir para excluir o cluster.