Serviço Gerenciado para Apache Spark

1. Visão geral do Serverless para Apache Spark

O Serviço gerenciado para Apache Spark é 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 Apache Spark gerenciado para modernização de data lake, ETL / ELT e ciência de dados segura em escala planetária. O Apache Spark gerenciado também é totalmente integrado a vários serviços do Google Cloud, incluindo BigQuery, Cloud Storage, mecanismo de agente do Gemini Enterprise e Knowledge Catalog.

O Apache Spark gerenciado está disponível em duas versões:

• O Apache Spark sem servidor gerenciado permite executar jobs do PySpark sem precisar configurar infraestrutura e escalonamento automático. O Apache Spark gerenciado oferece suporte a cargas de trabalho e sessões / notebooks em lote do PySpark.
• Os clusters gerenciados do Apache Spark permitem 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 na nuvem com o escalonamento vertical ou horizontal que quiser, incluindo o escalonamento automático.

Neste codelab, você vai aprender várias maneiras diferentes de usar o Dataproc sem servidor.

O Apache Spark foi originalmente criado para ser executado em clusters do Hadoop e usou o YARN como gerenciador de recursos. A manutenção de clusters Hadoop exige um conjunto específico de conhecimentos e a configuração adequada de vários parâmetros diferentes nos clusters. Além disso, o Spark também exige que o usuário defina um conjunto separado de controles. Isso leva a muitos cenários em que os desenvolvedores gastam mais tempo configurando a infraestrutura do que trabalhando no código do Spark.

O Dataproc sem servidor elimina a necessidade de configurar manualmente clusters do Hadoop ou do Spark. O Dataproc sem servidor não é executado no Hadoop e usa a própria alocação dinâmica de recursos para determinar os requisitos de recursos, incluindo o escalonamento automático. Um pequeno subconjunto de propriedades do Spark ainda pode ser personalizado com o Dataproc sem servidor, mas na maioria dos casos não é necessário fazer ajustes.

2. Configurar

Primeiro, você vai configurar seu ambiente e os recursos usados neste codelab.

Crie um projeto do Google Cloud. Você pode usar um já existente.

Clique em Cloud Shell na barra de ferramentas do Console do Cloud.

O Cloud Shell oferece um ambiente shell pronto para uso que pode ser usado neste codelab.

O Cloud Shell define o nome do projeto por padrão. Faça uma segunda verificação executando echo $GOOGLE_CLOUD_PROJECT. Se o ID do projeto não aparecer na saída, defina-o.

export GOOGLE_CLOUD_PROJECT=<your-project-id>

Defina uma região do Compute Engine para seus recursos, como us-central1 ou europe-west2.

export REGION=<your-region>

Ativar APIs

O codelab usa as seguintes APIs:

• BigQuery
• Dataproc

Ative as APIs necessárias. Isso leva cerca de um minuto, e uma mensagem de sucesso aparece quando o processo é concluído.

gcloud services enable bigquery.googleapis.com
gcloud services enable dataproc.googleapis.com

Configurar o acesso à rede

O Dataproc sem servidor exige que o Acesso particular do Google esteja ativado na região em que você vai executar os jobs do Spark, já que os drivers e executores do Spark têm apenas IPs particulares. Execute o seguinte comando para ativar na sub-rede default.

gcloud compute networks subnets update default \
  --region=${REGION} \
  --enable-private-ip-google-access

Para verificar se o Acesso privado do Google está ativado, execute o seguinte comando, que vai gerar True ou False.

gcloud compute networks subnets describe default \
  --region=${REGION} \
  --format="get(privateIpGoogleAccess)"

Criar um bucket de armazenamento

Crie um bucket de armazenamento que será usado para armazenar os recursos criados neste codelab.

Escolha um nome para o bucket. Os nomes de buckets precisam ser globalmente exclusivos para todos os usuários.

export BUCKET=<your-bucket-name>

Crie o bucket na região em que você pretende executar os jobs do Spark.

gsutil mb -l ${REGION} gs://${BUCKET}

É possível ver que seu bucket está disponível no console do Cloud Storage. Você também pode executar gsutil ls para ver seu bucket.

Crie um servidor de histórico permanente

A interface do Spark oferece um conjunto avançado de ferramentas de depuração e insights sobre jobs do Spark. Para conferir a interface do Spark dos jobs concluídos do Dataproc sem servidor, crie um cluster do Dataproc de nó único para usar como um servidor de histórico persistente.

Defina um nome para o servidor do histórico persistente.

PHS_CLUSTER_NAME=my-phs

Execute o comando a seguir.

gcloud dataproc clusters create ${PHS_CLUSTER_NAME} \
    --region=${REGION} \
    --single-node \
    --enable-component-gateway \
    --properties=spark:spark.history.fs.logDirectory=gs://${BUCKET}/phs/*/spark-job-history

A interface do Spark e o servidor de histórico permanente serão abordados em mais detalhes mais adiante no codelab.

3. Executar jobs do Spark sem servidor com lotes do Dataproc

Neste exemplo, você vai trabalhar com um conjunto de dados do conjunto de dados públicos do Citi Bike de Nova York (NYC). O NYC Citi Bikes é um sistema de aluguel de bicicletas pago em Nova York. Você vai fazer algumas transformações simples e imprimir os 10 IDs de estação mais famosos do Citi Bike. Este exemplo também usa o spark-bigquery-connector de código aberto para ler e gravar dados entre o Spark e o BigQuery sem problemas.

Clone o seguinte repositório do GitHub e cd no diretório que contém o arquivo citibike.py.

git clone https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos/data-analytics/next-2022-workshop/dataproc-serverless

citibike.py

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import BooleanType

if len(sys.argv) == 1:
    print("Please provide a GCS bucket name.")

bucket = sys.argv[1]
table = "bigquery-public-data:new_york_citibike.citibike_trips"

spark = SparkSession.builder \
          .appName("pyspark-example") \
          .config("spark.jars","gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar") \
          .getOrCreate()

df = spark.read.format("bigquery").load(table)

top_ten = df.filter(col("start_station_id") \
            .isNotNull()) \
            .groupBy("start_station_id") \
            .count() \
            .orderBy("count", ascending=False) \
            .limit(10) \
            .cache()

top_ten.show()

top_ten.write.option("header", True).csv(f"gs://{bucket}/citibikes_top_ten_start_station_ids")

Envie o job para o Spark sem servidor usando o SDK do Cloud, disponível no Cloud Shell por padrão. Execute o comando a seguir no shell, que usa o SDK Cloud e a API Dataproc Batches para enviar jobs do Spark sem servidor.

gcloud dataproc batches submit pyspark citibike.py \
  --batch=citibike-job \
  --region=${REGION} \
  --deps-bucket=gs://${BUCKET} \
  --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar \
--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER_NAME} \
  -- ${BUCKET}

Para explicar melhor:

• gcloud dataproc batches submit faz referência à API Dataproc Batches.
• pyspark indica que você está enviando um job do PySpark.
• --batch é o nome do job. Se não for informado, um UUID gerado aleatoriamente será usado.
• --region=${REGION} é a região geográfica em que o job será processado.
• --deps-bucket=${BUCKET} é onde o arquivo Python local é enviado por upload antes de ser executado no ambiente sem servidor.
• O --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar inclui o jar do spark-bigquery-connector no ambiente de execução do Spark.
• --history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER} é o nome totalmente qualificado do servidor de histórico permanente. É aqui que os dados de eventos do Spark (separados da saída do console) são armazenados e podem ser visualizados na interface do Spark.
• O -- final indica que qualquer coisa além disso será argumentos de tempo de execução para o programa. Nesse caso, você está enviando o nome do bucket, conforme exigido pelo job.

A saída a seguir vai aparecer quando o lote for enviado.

Batch [citibike-job] submitted.

Após alguns minutos, você verá a saída a seguir com os metadados do job.

+----------------+------+
|start_station_id| count|
+----------------+------+
|             519|551078|
|             497|423334|
|             435|403795|
|             426|384116|
|             293|372255|
|             402|367194|
|             285|344546|
|             490|330378|
|             151|318700|
|             477|311403|
+----------------+------+

Batch [citibike-job] finished.

Na próxima seção, você vai aprender a localizar os registros desse job.

Recursos adicionais

Com o Spark sem servidor, você tem mais opções para executar seus jobs.

• É possível criar uma imagem do Docker personalizada para executar seu job. Essa é uma ótima maneira de incluir dependências adicionais, incluindo bibliotecas Python e R.
• É possível conectar uma instância do Metastore do Dataproc ao seu job para acessar os metadados do Hive.
• Para ter mais controle, o Dataproc Serverless oferece suporte à configuração de um pequeno conjunto de propriedades do Spark.

4. Métricas e observabilidade do Dataproc

O console do Dataproc Batches lista todos os seus jobs do Dataproc sem servidor. No console, você encontra o ID do lote, o local, o status, o horário de criação, o tempo decorrido e o tipo de cada job. Clique no ID do lote do job para ver mais informações sobre ele.

Nesta página, você verá informações como Monitoramento, que mostra quantos executores do Spark em lote seu job usou ao longo do tempo (indicando o quanto ele foi escalonado automaticamente).

Na guia Detalhes, você encontra mais metadados sobre o job, incluindo argumentos e parâmetros enviados com ele.

Também é possível acessar todos os registros nessa página. Quando os jobs do Dataproc sem servidor são executados, três conjuntos diferentes de registros são gerados:

• No nível de serviço
• Saída do console
• Geração de registros de eventos do Spark

Nível de serviço: inclui registros gerados pelo serviço sem servidor do Dataproc, como solicitações de CPUs extras para escalonamento automático. Para ver esses registros, clique em Ver registros, que vai abrir o Cloud Logging.

A saída do console pode ser vista em Saída.Essa é a saída gerada pelo job, incluindo metadados que o Spark imprime ao iniciar um job ou qualquer instrução de impressão incorporada a ele.

O registro de eventos do Spark pode ser acessado na interface do Spark. Como você forneceu ao job do Spark um servidor de histórico persistente, é possível acessar a interface do Spark clicando em Acessar o servidor de histórico do Spark, que contém informações sobre os jobs do Spark executados anteriormente. Saiba mais sobre a interface do Spark na documentação oficial do Spark.

5. Modelos do Dataproc: BQ -> GCS

Os modelos do Dataproc são ferramentas de código aberto que ajudam a simplificar ainda mais as tarefas de processamento de dados na nuvem. Eles servem como um wrapper para o Dataproc Serverless e incluem modelos para muitas tarefas de importação e exportação de dados, incluindo:

• BigQuerytoGCS e GCStoBigQuery
• GCStoBigTable
• GCStoJDBC e JDBCtoGCS
• HivetoBigQuery
• MongotoGCS e GCStoMongo

A lista completa está disponível no README.

Nesta seção, você vai usar os modelos do Dataproc para exportar dados do BigQuery para o GCS.

Clone o repositório

Clone o repositório e mude para a pasta python.

git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git
cd dataproc-templates/python

Configurar o ambiente

Agora você vai definir variáveis de ambiente. Os modelos do Dataproc usam a variável de ambiente GCP_PROJECT para o ID do projeto. Portanto, defina-a como GOOGLE_CLOUD_PROJECT..

export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}

Sua região precisa ser definida no ambiente de antes. Caso contrário, defina aqui.

export REGION=<region>

Os modelos do Dataproc usam o spark-bigquery-connector para processar jobs do BigQuery e exigem que o URI seja incluído em uma variável de ambiente JARS. Defina a variável JARS.

export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"

Configurar parâmetros de modelo

Defina o nome de um bucket de staging para o serviço usar.

export GCS_STAGING_LOCATION=gs://${BUCKET}

Em seguida, você vai definir algumas variáveis específicas do job. Para a tabela de entrada, você vai referenciar novamente o conjunto de dados do BigQuery NYC Citibike.

BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips

Você pode escolher csv, parquet, avro ou json. Para este codelab, escolha CSV. Na próxima seção, você vai aprender a usar os modelos do Dataproc para converter tipos de arquivo.

BIGQUERY_GCS_OUTPUT_FORMAT=csv

Defina o modo de saída como overwrite. Você pode escolher entre overwrite, append, ignore ou errorifexists.

BIGQUERY_GCS_OUTPUT_MODE=overwrite

Defina o local de saída do GCS como um caminho no bucket.

BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS

Executar o modelo

Execute o modelo BIGQUERYTOGCS especificando-o abaixo e fornecendo os parâmetros de entrada definidos.

./bin/start.sh \
-- --template=BIGQUERYTOGCS \
        --bigquery.gcs.input.table=${BIGQUERY_GCS_INPUT_TABLE} \
        --bigquery.gcs.output.format=${BIGQUERY_GCS_OUTPUT_FORMAT} \
        --bigquery.gcs.output.mode=${BIGQUERY_GCS_OUTPUT_MODE} \
        --bigquery.gcs.output.location=${BIGQUERY_GCS_OUTPUT_LOCATION}

A saída será bastante ruidosa, mas depois de cerca de um minuto, você verá o seguinte.

Batch [5766411d6c78444cb5e80f305308d8f8] submitted.
...
Batch [5766411d6c78444cb5e80f305308d8f8] finished.

Para verificar se os arquivos foram gerados, execute o seguinte comando:

gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}

Por padrão, o Spark grava em vários arquivos, dependendo da quantidade de dados. Nesse caso, você vai encontrar cerca de 30 arquivos gerados. Os nomes dos arquivos de saída do Spark são formatados com part, seguido por um número de cinco dígitos (indicando o número da peça) e uma string hash. Para grandes quantidades de dados, o Spark geralmente grava em vários arquivos. Um exemplo de nome de arquivo é part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv.

6. Modelos do Dataproc: CSV para Parquet

Agora você vai usar os modelos do Dataproc para converter dados no GCS de um tipo de arquivo para outro usando o GCSTOGCS. Esse modelo usa o SparkSQL e oferece a opção de enviar uma consulta do SparkSQL para ser processada durante a transformação para processamento adicional.

Confirmar variáveis de ambiente

Confirme se GCP_PROJECT, REGION e GCS_STAGING_BUCKET estão definidos na seção anterior.

echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}

Definir parâmetros do modelo

Agora, defina os parâmetros de configuração para GCStoGCS. Comece com o local dos arquivos de entrada. Observe que este é um diretório e não um arquivo específico, já que todos os arquivos no diretório serão processados. Defina como BIGQUERY_GCS_OUTPUT_LOCATION.

GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}

Defina o formato do arquivo de entrada.

GCS_TO_GCS_INPUT_FORMAT=csv

Defina o formato de saída desejado. Você pode escolher parquet, json, avro ou csv.

GCS_TO_GCS_OUTPUT_FORMAT=parquet

Defina o modo de saída como overwrite. Você pode escolher entre overwrite, append, ignore ou errorifexists.

GCS_TO_GCS_OUTPUT_MODE=overwrite

Defina o local de saída.

GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS

Executar o modelo

Execute o modelo GCStoGCS.

./bin/start.sh \
-- --template=GCSTOGCS \
        --gcs.to.gcs.input.location=${GCS_TO_GCS_INPUT_LOCATION} \
        --gcs.to.gcs.input.format=${GCS_TO_GCS_INPUT_FORMAT} \
        --gcs.to.gcs.output.format=${GCS_TO_GCS_OUTPUT_FORMAT} \
        --gcs.to.gcs.output.mode=${GCS_TO_GCS_OUTPUT_MODE} \
        --gcs.to.gcs.output.location=${GCS_TO_GCS_OUTPUT_LOCATION}

A saída será bastante ruidosa, mas depois de cerca de um minuto, você vai ver uma mensagem de sucesso como a abaixo.

Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted.
...
Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.

Para verificar se os arquivos foram gerados, execute o seguinte comando:

gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}

Com esse modelo, você também tem a opção de fornecer consultas do SparkSQL transmitindo gcs.to.gcs.temp.view.name e gcs.to.gcs.sql.query para o modelo, permitindo que uma consulta do SparkSQL seja executada nos dados antes da gravação no GCS.

7. Limpar recursos

Para evitar cobranças desnecessárias na sua conta do GCP após a conclusão deste codelab:

1. Exclua o bucket do Cloud Storage do ambiente que você criou.

gsutil rm -r gs://${BUCKET}

2. Exclua o cluster do Dataproc usado para o servidor de histórico persistente.

gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
  --region=${REGION}

3. Exclua os jobs sem servidor do Dataproc. Acesse o console de lotes, clique na caixa ao lado de cada job que você quer excluir e clique em EXCLUIR.

Se você criou um projeto apenas para este codelab, também é possível excluir o projeto:

1. No Console do GCP, acesse a página Projetos.
2. Na lista de projetos, selecione o que você quer excluir e clique em "Excluir".
3. Na caixa, digite o ID do projeto e clique em "Encerrar" para excluí-lo.

Cuidado:excluir um projeto tem os seguintes efeitos:

• Tudo no projeto é excluído. Se você tiver usado um projeto atual para seguir este tutorial, a exclusão dele incluirá quaisquer outras ações feitas no projeto.
• Os IDs do projeto personalizados são perdidos. Ao criar o projeto, você pode ter criado um ID do projeto personalizado para ser usado no futuro. Para preservar os URLs que usam o ID do projeto, como um URL appspot.com, exclua recursos específicos do projeto, em vez de excluir o projeto inteiro.

8. A seguir

Os recursos a seguir oferecem outras maneiras de aproveitar o Serverless Spark:

• Saiba como orquestrar fluxos de trabalho do Dataproc sem servidor usando o Cloud Composer.
• Saiba como integrar o Dataproc sem servidor aos pipelines do Kubeflow.