O Apache Kafka é uma plataforma de streaming distribuída de código aberto para pipelines de dados em tempo real e integração de dados. Ele oferece um sistema de streaming eficiente e escalonável para uso em diversos aplicativos, incluindo:
- Análise em tempo real
- Processamento de stream
- Agregação de registros
- Mensagens distribuídas
- Streaming de eventos
Objetivos
Instalar o Kafka em um cluster de alta disponibilidade do Dataproc com o ZooKeeper (chamado neste tutorial de "cluster do Dataproc Kafka").
Criar dados fictícios de clientes e depois publicá-los em um tópico do Kafka.
Criar tabelas em parquet e ORC do Hive no Cloud Storage para receber dados de tópicos do Kafka transmitidos por streaming.
Envie um job do PySpark para se inscrever e fazer streaming do tópico do Kafka para o Cloud Storage nos formatos parquet e ORC.
Execute uma consulta nos dados da tabela Hive transmitida para contar as mensagens do Kafka transmitidas.
Custos
Neste documento, você usará os seguintes componentes faturáveis do Google Cloud:
Para gerar uma estimativa de custo baseada na projeção de uso deste tutorial, use a calculadora de preços.
Ao concluir as tarefas descritas neste documento, é possível evitar o faturamento contínuo excluindo os recursos criados. Saiba mais em Limpeza.
Antes de começar
Crie um projeto do Google Cloud, caso ainda não tenha feito isso.
- Faça login na sua conta do Google Cloud. Se você começou a usar o Google Cloud agora, crie uma conta para avaliar o desempenho de nossos produtos em situações reais. Clientes novos também recebem US$ 300 em créditos para executar, testar e implantar cargas de trabalho.
-
No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.
-
Verifique se a cobrança está ativada para o seu projeto do Google Cloud.
-
Ative as APIs Dataproc, Compute Engine, and Cloud Storage.
-
No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.
-
Verifique se a cobrança está ativada para o seu projeto do Google Cloud.
-
Ative as APIs Dataproc, Compute Engine, and Cloud Storage.
- No console do Cloud, acesse a página Buckets do Cloud Storage.
- Clique em Criar bucket.
- Na página Criar um bucket, insira as informações do seu bucket. Para ir à próxima
etapa, clique em Continuar.
- Em Nomear o bucket, insira um nome que atenda aos requisitos de nomenclatura de bucket.
-
Em Escolha um local para armazenar seus dados, faça o seguinte:
- Selecione uma opção de Tipo de local.
- Escolha uma opção de Local.
- Em Escolher uma classe de armazenamento padrão para seus dados, selecione uma classe de armazenamento.
- Em Escolha como controlar o acesso a objetos, selecione uma opção de Controle de acesso.
- Em Configurações avançadas (opcional), especifique um método de criptografia, uma política de retenção ou rótulos de bucket.
- Clique em Criar.
Etapas do tutorial
Siga as etapas abaixo para criar um cluster do Dataproc Kafka e ler um tópico do Kafka no Cloud Storage no formato parquet OU ORC.
Copie o script de instalação do Kafka no Cloud Storage
O script de ação de inicialização kafka.sh
instala o Kafka em um cluster do Dataproc.
Procure o código.
Copie o script de ação de inicialização
kafka.sh
para o bucket do Cloud Storage. O script instala o Kafka em um cluster do Dataproc.Abra o Cloud Shell e execute o seguinte comando:
gsutil cp gs://goog-dataproc-initialization-actions-REGION/kafka/kafka.sh gs://BUCKET_NAME/scripts/
Faça as seguintes substituições:
- REGION:
kafka.sh
é armazenado em buckets públicos com tags regionais no Cloud Storage. Especifique uma região do Compute Engine geograficamente próxima (exemplo:us-central1
). - BUCKET_NAME: o nome do bucket do Cloud Storage.
- REGION:
Criar um cluster do Dataproc Kafka
Abra o Cloud Shell e execute o seguinte comando
gcloud dataproc clusters create
para criar um cluster de alta disponibilidade do Dataproc que instala os componentes do Kafka e do ZooKeeper:gcloud dataproc clusters create KAFKA_CLUSTER \ --project=PROJECT_ID \ --region=REGION \ --image-version=2.1-debian11 \ --num-masters=3 \ --enable-component-gateway \ --initialization-actions=gs://BUCKET_NAME/scripts/kafka.sh
Observações:
- KAFKA_CLUSTER: o nome do cluster, que precisa ser exclusivo em um projeto. O nome precisa começar com uma letra minúscula e pode conter até 51 letras minúsculas, números e hifens. Ele não pode terminar com hífen. O nome de um cluster excluído pode ser reutilizado.
- PROJECT_ID: o projeto a ser associado a esse cluster.
- REGION: a
região do Compute Engine
em que o cluster estará localizado, como
us-central1
.- É possível adicionar a sinalização
--zone=ZONE
opcional para especificar uma zona dentro da região especificada, comous-central1-a
. Se você não especificar uma zona, o recurso de colocação em zona automática do Dataproc selecionará uma zona com a região especificada.
- É possível adicionar a sinalização
--image-version
: a versão de imagem2.1-debian11
do Dataproc é recomendada para este tutorial. Observação: cada versão de imagem contém um conjunto de componentes pré-instalados, incluindo o componente Hive usado neste tutorial (consulte Versões de imagem do Dataproc compatíveis).--num-master
:3
nós mestres criam um cluster de alta disponibilidade. O componente Zookeeper, que é exigido pelo Kafka, é pré-instalado em um cluster de alta disponibilidade.--enable-component-gateway
: ativa o Gateway de componentes do Dataproc.- BUCKET_NAME: o nome do bucket do Cloud Storage
que contém o script de inicialização
/scripts/kafka.sh
(consulte Copiar o script de instalação do Kafka para o Cloud Storage).
Crie um tópico custdata
do Kafka
Para criar um tópico do Kafka no cluster do Kafka do Dataproc:
Use o utilitário SSH para abrir uma janela do terminal na VM mestre do cluster.
Crie um tópico
custdata
do Kafka./usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --create --topic custdata
Observações:
KAFKA_CLUSTER: insira o nome do cluster do Kafka.
-w-0:9092
significa o agente do Kafka em execução na porta9092
no nóworker-0
.É possível executar os seguintes comandos depois de criar o tópico
custdata
:# List all topics. /usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --list
# Consume then display topic data. /usr/lib/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --topic custdata
# Count the number of messages in the topic. /usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata # Delete topic. /usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --delete --topic custdata
Publicar conteúdo no tópico custdata
do Kafka
O script a seguir usa a ferramenta kafka-console-producer.sh
Kafka para gerar dados fictícios do cliente no formato CSV.
Copie e cole o script no terminal SSH do nó mestre do cluster do Kafka. Pressione <return> para executar o script.
for i in {1..10000}; do \ custname="cust name${i}" uuid=$(dbus-uuidgen) age=$((45 + $RANDOM % 45)) amount=$(echo "$(( $RANDOM % 99999 )).$(( $RANDOM % 99 ))") message="${uuid}:${custname},${age},${amount}" echo ${message} done | /usr/lib/kafka/bin/kafka-console-producer.sh \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata \ --property "parse.key=true" \ --property "key.separator=:"
Observações:
- KAFKA_CLUSTER: o nome do cluster do Kafka.
Execute o seguinte comando do Kafka para confirmar se o tópico
custdata
contém 10.000 mensagens./usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata
Observações:
- KAFKA_CLUSTER: o nome do cluster do Kafka.
Resposta esperada:
custdata:0:10000
Criar tabelas do Hive no Cloud Storage
Criar tabelas do Hive para receber dados do tópico do Kafka transmitidos por streaming.
Execute as etapas a seguir para criar as tabelas Hive cust_parquet
(parquet) e cust_orc
(ORC) no bucket do Cloud Storage.
Insira o BUCKET_NAME no script a seguir, copie e cole o script no terminal SSH do nó mestre do cluster do Kafka e pressione <return> para criar um script da
~/hivetables.hql
(linguagem de consulta do Hive).Você executará o script
~/hivetables.hql
na próxima etapa para criar tabelas em parquet e ORC Hive no bucket do Cloud Storage.cat > ~/hivetables.hql <<EOF drop table if exists cust_parquet; create external table if not exists cust_parquet (uuid string, custname string, age string, amount string) row format delimited fields terminated by ',' stored as parquet location "gs://BUCKET_NAME/tables/cust_parquet"; drop table if exists cust_orc; create external table if not exists cust_orc (uuid string, custname string, age string, amount string) row format delimited fields terminated by ',' stored as orc location "gs://BUCKET_NAME/tables/cust_orc"; EOF
No terminal SSH do nó mestre do cluster do Kafka, envie o job
~/hivetables.hql
do Hive para criar as tabelascust_parquet
(parquet) ecust_orc
(ORC) do Hive no bucket do Cloud Storage.gcloud dataproc jobs submit hive \ --cluster=KAFKA_CLUSTER \ --region=REGION \ -f ~/hivetables.hql
Observações:
- O componente Hive é pré-instalado no cluster do Kafka do Dataproc. Consulte Versões de lançamento 2.1.x para uma lista das versões do componente Hive incluídas nas imagens 2.1 lançadas recentemente.
- KAFKA_CLUSTER: o nome do cluster do Kafka.
- REGION: a região em que o cluster do Kafka está localizado.
Fazer streaming do custdata
do Kafka para tabelas do Hive
- Execute o comando a seguir no terminal SSH do nó mestre do
cluster do Kafka para instalar a biblioteca
kafka-python
. Um cliente Kafka é necessário para fazer streaming de dados do tópico Kafka para o Cloud Storage.
pip install kafka-python
Insira o BUCKET_NAME, copie e cole o código PySpark a seguir no terminal SSH do nó mestre do cluster do Kafka e pressione <return> para criar um arquivo
streamdata.py
.O script assina o tópico
custdata
do Kafka e transmite os dados para as tabelas do Hive no Cloud Storage. O formato de saída, que pode ser parquet ou ORC, é transmitido para o script como um parâmetro.cat > streamdata.py <<EOF #!/bin/python import sys from pyspark.sql.functions import * from pyspark.sql.types import * from pyspark.sql import SparkSession from kafka import KafkaConsumer def getNameFn (data): return data.split(",")[0] def getAgeFn (data): return data.split(",")[1] def getAmtFn (data): return data.split(",")[2] def main(cluster, outputfmt): spark = SparkSession.builder.appName("APP").getOrCreate() spark.sparkContext.setLogLevel("WARN") Logger = spark._jvm.org.apache.log4j.Logger logger = Logger.getLogger(__name__) rows = spark.readStream.format("kafka") \ .option("kafka.bootstrap.servers", cluster+"-w-0:9092").option("subscribe", "custdata") \ .option("startingOffsets", "earliest")\ .load() getNameUDF = udf(getNameFn, StringType()) getAgeUDF = udf(getAgeFn, StringType()) getAmtUDF = udf(getAmtFn, StringType()) logger.warn("Params passed in are cluster name: " + cluster + " output format(sink): " + outputfmt) query = rows.select (col("key").cast("string").alias("uuid"),\ getNameUDF (col("value").cast("string")).alias("custname"),\ getAgeUDF (col("value").cast("string")).alias("age"),\ getAmtUDF (col("value").cast("string")).alias("amount")) writer = query.writeStream.format(outputfmt)\ .option("path","gs://BUCKET_NAME/tables/cust_"+outputfmt)\ .option("checkpointLocation", "gs://BUCKET_NAME/chkpt/"+outputfmt+"wr") \ .outputMode("append")\ .start() writer.awaitTermination() if __name__=="__main__": if len(sys.argv) < 2: print ("Invalid number of arguments passed ", len(sys.argv)) print ("Usage: ", sys.argv[0], " cluster format") print ("e.g.: ", sys.argv[0], " <cluster_name> orc") print ("e.g.: ", sys.argv[0], " <cluster_name> parquet") main(sys.argv[1], sys.argv[2]) EOF
No terminal SSH do nó mestre do cluster do Kafka, execute
spark-submit
para transmitir dados para as tabelas do Hive no Cloud Storage.Insira o nome do KAFKA_CLUSTER e a saída FORMAT. Em seguida, copie e cole o código a seguir no terminal SSH do nó mestre do cluster do Kafka e pressione <return> para executar o código e fazer streaming dos dados
custdata
do Kafka em formato parquet para as tabelas do Hive no Cloud Storage.spark-submit --packages \ org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.3,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 \ --conf spark.history.fs.gs.outputstream.type=FLUSHABLE_COMPOSITE \ --conf spark.driver.memory=4096m \ --conf spark.executor.cores=2 \ --conf spark.executor.instances=2 \ --conf spark.executor.memory=6144m \ streamdata.py KAFKA_CLUSTER FORMAT
Observações:
- KAFKA_CLUSTER: insira o nome do cluster do Kafka.
- FORMAT: especifique
parquet
ouorc
como o formato de saída. É possível executar o comando sucessivamente para transmitir os dois formatos para as tabelas do Hive: por exemplo, na primeira invocação, especifiqueparquet
para transmitir o tópicocustdata
do Kafka para a tabela do Parce do Hive. Depois, na segunda invocação, especifique o formatoorc
para transmitircustdata
para a tabela ORC do Hive.
Depois que a saída padrão for interrompida no terminal SSH, o que significa que todo o
custdata
foi transmitido, pressione <control-c> no terminal SSH para interromper o processo.Listar as tabelas do Hive no Cloud Storage.
gsutil ls -r gs://BUCKET_NAME/tables/*
Observações:
- BUCKET_NAME: insira o nome do bucket do Cloud Storage que contém as tabelas do Hive. Consulte Criar tabelas do Hive.
Consultar dados transmitidos
No terminal SSH do nó mestre do cluster do Kafka, execute o seguinte comando
hive
para contar as mensagenscustdata
do Kafka transmitidas nas tabelas do Hive no Cloud Storage.hive -e "select count(1) from TABLE_NAME"
Observações:
- TABLE_NAME: especifique
cust_parquet
oucust_orc
como o nome da tabela do Hive.
Snippet de saída esperado:
- TABLE_NAME: especifique
...
Status: Running (Executing on YARN cluster with App id application_....)
----------------------------------------------------------------------------------------------
VERTICES MODE STATUS TOTAL COMPLETED RUNNING PENDING FAILED KILLED
----------------------------------------------------------------------------------------------
Map 1 .......... container SUCCEEDED 1 1 0 0 0 0
Reducer 2 ...... container SUCCEEDED 1 1 0 0 0 0
----------------------------------------------------------------------------------------------
VERTICES: 02/02 [==========================>>] 100% ELAPSED TIME: 9.89 s
----------------------------------------------------------------------------------------------
OK
10000
Time taken: 21.394 seconds, Fetched: 1 row(s)
Limpar
Exclua o projeto
Exclua um projeto do Google Cloud:
gcloud projects delete PROJECT_ID
Excluir recursos
-
Excluir o bucket:
gcloud storage buckets delete BUCKET_NAME
- Exclua o cluster do Kafka:
gcloud dataproc clusters delete KAFKA_CLUSTER \ --region=${REGION}