O Apache Kafka é uma plataforma de streaming distribuída de código aberto para 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
Instale o Kafka em uma Cluster de alta disponibilidade do Dataproc com o ZooKeeper (referido neste tutorial como "cluster Kafka do Dataproc").
Criar dados fictícios de clientes e publicá-los em um tópico do Kafka.
Criar tabelas em parquet Hive e ORC no Cloud Storage para receber dados transmitidos de tópicos do Kafka.
Envie um job do PySpark para fazer a assinatura e transmitir o tópico do Kafka para o Cloud Storage nos formatos parquet e ORC.
Executar uma consulta nos dados da tabela Hive transmitida para contar a transmissão Mensagens do Kafka.
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 Kafka do Dataproc e ler um tópico Kafka no Cloud Storage no formato parquet OU ORC;
Copie o script de instalação do Kafka para o Cloud Storage
A ação de inicialização kafka.sh
instala o Kafka em um cluster do Dataproc.
Procure o código.
Copiar
kafka.sh
ação de inicialização no seu bucket do Cloud Storage. O script instala o Kafka em um cluster do Dataproc.Abra o Cloud Shell e execute o seguinte comando:
gcloud storage 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 tags públicas com tag regional no Cloud Storage. Especificar um Região do Compute Engine, Por exemplo,us-central1
. - BUCKET_NAME: o nome do bucket do Cloud Storage.
- REGION:
crie um cluster Kafka do Dataproc
Abra o Cloud Shell e execute o seguinte
gcloud dataproc clusters create
para criar um bucket do Dataproc Cluster de alta disponibilidade cluster que instala os componentes Kafka e 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. letras, números e hifens. Ele não pode terminar com um hífen. O nome de um que o cluster excluído possa ser reutilizado.
- PROJECT_ID: o projeto a ser associado a este cluster.
- REGION: o
Região do Compute Engine
onde o cluster será 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, a Colocação em zona automática do Dataproc seleciona uma zona com a região especificada.
- É possível adicionar a sinalização
--image-version
: versão da imagem do Dataproc2.1-debian11
é recomendado para este tutorial. Observação: cada versão de imagem contém um conjunto de imagens componentes, 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, 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 no Cloud Storage.
Criar um tópico custdata
do Kafka
Para criar um tópico Kafka no cluster Kafka do Dataproc:
Use o SSH para abrir uma janela de terminal na VM mestre do cluster.
Criar 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 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 kafka-console-producer.sh
para
gerar dados fictícios de clientes no formato CSV.
Copie e cole o script no SSH terminal no 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 comando Kafka a seguir 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
Crie tabelas do Hive no Cloud Storage
Criar tabelas do Hive para receber dados transmitidos do tópico do Kafka.
Siga as etapas abaixo para criar cust_parquet
(parquet) e um
Tabelas Hive cust_orc
(ORC) no bucket do Cloud Storage.
Insira seu BUCKET_NAME no script a seguir, depois copie e cole o script no terminal SSH do nó mestre do cluster do Kafka, e pressione <return> para criar um script
~/hivetables.hql
(linguagem de consulta do Hive).Você vai 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 Kafka, envie o job
~/hivetables.hql
do Hive para criar tabelas do Hivecust_parquet
(parquet) e docust_orc
(ORC) no bucket do Cloud Storage.gcloud dataproc jobs submit hive \ --cluster=KAFKA_CLUSTER \ --region=REGION \ -f ~/hivetables.hql
Observações:
- O componente Hive vem pré-instalado no Kafka do Dataproc aglomerado. 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.
Transmitir custdata
do Kafka para tabelas do Hive
- Execute o seguinte comando no terminal SSH do nó mestre de
cluster Kafka para instalar a biblioteca
kafka-python
. Um cliente Kafka é necessário para transmitir dados de tópicos do Kafka para Cloud Storage.
pip install kafka-python
Insira seu BUCKET_NAME, depois copie e cole o seguinte o código do PySpark para o terminal SSH do nó mestre do cluster do Kafka e pressione <return> para criar um arquivo
streamdata.py
.O script se inscreve no tópico
custdata
do Kafka e transmite a às tabelas 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 de cluster Kafka, execute
spark-submit
para fazer streaming de dados para das tabelas do Hive no Cloud Storage.Insira o nome da KAFKA_CLUSTER e a saída FORMAT. Depois copie e cole o seguinte código no SSH terminal no nó mestre do cluster do Kafka e pressione <return>. para executar o código e transmitir os dados
custdata
do Kafka em formato parquet para 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 de forma sucessiva para transmitir ambos os formatos para as tabelas Hive: por exemplo, na primeira invocação, especifiqueparquet
para transmitir o tópicocustdata
do Kafka para o Hive mesa parquet; e, na segunda invocação, especifica o formatoorc
para streamcustdata
para a tabela ORC do Hive.
Depois que a saída padrão parar no terminal SSH, o que significa que todo o conteúdo de
custdata
foi transmitido, pressione <control-c> no terminal SSH para interromper o processo.Listar as tabelas do Hive no Cloud Storage.
gcloud storage ls gs://BUCKET_NAME/tables/* --recursive
Observações:
- BUCKET_NAME: insira o nome do Cloud Storage bucket que contém suas tabelas do Hive (consulte Criar tabelas do Hive).
Consultar dados de streaming
No terminal SSH do nó mestre de seu cluster Kafka, execute o seguinte comando
hive
para contar as mensagens transmitidas decustdata
do Kafka nas tabelas 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 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 Kafka:
gcloud dataproc clusters delete KAFKA_CLUSTER \ --region=${REGION}