Transmitir um tópico do Kafka para o Hive

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 vários aplicativos, incluindo:

  • Análise em tempo real
  • Processamento de stream
  • Agregação de registros
  • Mensagens distribuídas
  • Streaming de eventos

Objetivos

  1. Instale o Kafka em um cluster HA do Dataproc com o ZooKeeper (neste tutorial, chamado de "cluster do Kafka do Dataproc").

  2. Crie dados fictícios de clientes e publique-os em um tópico do Kafka.

  3. Crie tabelas do Hive parquet e ORC no Cloud Storage para receber dados de tópicos do Kafka transmitidos.

  4. Envie um job do PySpark para se inscrever e transmitir o tópico do Kafka para o Cloud Storage nos formatos Parquet e ORC.

  5. Execute uma consulta nos dados da tabela do Hive transmitidos para contar as mensagens do Kafka transmitidas.

Custos

Neste documento, você vai 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.

Novos usuários do Google Cloud podem estar qualificados para uma avaliação gratuita.

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

Se você ainda não tiver um projeto, crie um Google Cloud .

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

  5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

    8.
  8. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets

  9. Click Create.
  10. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
    1. In the Get started section, do the following:
      • Enter a globally unique name that meets the bucket naming requirements.
      • To add a bucket label, expand the Labels section (), click Add label, and specify a key and a value for your label.
    2. In the Choose where to store your data section, do the following:
      1. Select a Location type.
      2. Choose a location where your bucket's data is permanently stored from the Location type drop-down menu.
      3. To set up cross-bucket replication, select Add cross-bucket replication via Storage Transfer Service and follow these steps:

        Set up cross-bucket replication

        1. In the Bucket menu, select a bucket.

        2. In the Replication settings section, click Configure to configure settings for the replication job.

          The Configure cross-bucket replication pane appears.

          • To filter objects to replicate by object name prefix, enter a prefix that you want to include or exclude objects from, then click Add a prefix.
          • To set a storage class for the replicated objects, select a storage class from the Storage class menu. If you skip this step, the replicated objects will use the destination bucket's storage class by default.
          • Click Done.
    3. In the Choose how to store your data section, do the following:
      1. Select a default storage class for the bucket or Autoclass for automatic storage class management of your bucket's data.
      2. To enable hierarchical namespace, in the Optimize storage for data-intensive workloads section, select Enable hierarchical namespace on this bucket.
    4. In the Choose how to control access to objects section, select whether or not your bucket enforces public access prevention, and select an access control method for your bucket's objects.
    5. In the Choose how to protect object data section, do the following:
      • Select any of the options under Data protection that you want to set for your bucket.
        • To enable soft delete, click the Soft delete policy (For data recovery) checkbox, and specify the number of days you want to retain objects after deletion.
        • To set Object Versioning, click the Object versioning (For version control) checkbox, and specify the maximum number of versions per object and the number of days after which the noncurrent versions expire.
        • To enable the retention policy on objects and buckets, click the Retention (For compliance) checkbox, and then do the following:
          • To enable Object Retention Lock, click the Enable object retention checkbox.
          • To enable Bucket Lock, click the Set bucket retention policy checkbox, and choose a unit of time and a length of time for your retention period.
      • To choose how your object data will be encrypted, expand the Data encryption section (), and select a Data encryption method.
  11. Click Create.

    12. Etapas do tutorial

    Siga estas etapas para criar um cluster do Dataproc Kafka e ler um tópico do Kafka no Cloud Storage no formato Parquet OU ORC.

    Copiar o script de instalação do Kafka para o Cloud Storage

    O script de ação de inicialização kafka.sh instala o Kafka em um cluster do Dataproc.

    1. Navegue pelo código. 

      #!/bin/bash
#    Copyright 2015 Google, Inc.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#
# This script installs Apache Kafka (http://kafka.apache.org) on a Google Cloud
# Dataproc cluster.

set -euxo pipefail

readonly ZOOKEEPER_HOME=/usr/lib/zookeeper
readonly KAFKA_HOME=/usr/lib/kafka
readonly KAFKA_PROP_FILE='/etc/kafka/conf/server.properties'
readonly ROLE="$(/usr/share/google/get_metadata_value attributes/dataproc-role)"
readonly RUN_ON_MASTER="$(/usr/share/google/get_metadata_value attributes/run-on-master || echo false)"
readonly KAFKA_ENABLE_JMX="$(/usr/share/google/get_metadata_value attributes/kafka-enable-jmx || echo false)"
readonly KAFKA_JMX_PORT="$(/usr/share/google/get_metadata_value attributes/kafka-jmx-port || echo 9999)"
readonly INSTALL_KAFKA_PYTHON="$(/usr/share/google/get_metadata_value attributes/install-kafka-python || echo false)"

# The first ZooKeeper server address, e.g., "cluster1-m-0:2181".
ZOOKEEPER_ADDRESS=''
# Integer broker ID of this node, e.g., 0
BROKER_ID=''

function retry_apt_command() {
  cmd="$1"
  for ((i = 0; i < 10; i++)); do
    if eval "$cmd"; then
      return 0
    fi
    sleep 5
  done
  return 1
}

function recv_keys() {
  retry_apt_command "apt-get install -y gnupg2 &&\
                     apt-key adv --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C"
}

function update_apt_get() {
  retry_apt_command "apt-get update"
}

function install_apt_get() {
  pkgs="$@"
  retry_apt_command "apt-get install -y $pkgs"
}

function err() {
  echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')]: $@" >&2
  return 1
}

# Returns the list of broker IDs registered in ZooKeeper, e.g., " 0, 2, 1,".
function get_broker_list() {
  ${KAFKA_HOME}/bin/zookeeper-shell.sh "${ZOOKEEPER_ADDRESS}" \
    <<<"ls /brokers/ids" |
    grep '\[.*\]' |
    sed 's/\[/ /' |
    sed 's/\]/,/'
}

# Waits for zookeeper to be up or time out.
function wait_for_zookeeper() {
  for i in {1..20}; do
    if "${ZOOKEEPER_HOME}/bin/zkCli.sh" -server "${ZOOKEEPER_ADDRESS}" ls /; then
      return 0
    else
      echo "Failed to connect to ZooKeeper ${ZOOKEEPER_ADDRESS}, retry ${i}..."
      sleep 5
    fi
  done
  echo "Failed to connect to ZooKeeper ${ZOOKEEPER_ADDRESS}" >&2
  exit 1
}

# Wait until the current broker is registered or time out.
function wait_for_kafka() {
  for i in {1..20}; do
    local broker_list=$(get_broker_list || true)
    if [[ "${broker_list}" == *" ${BROKER_ID},"* ]]; then
      return 0
    else
      echo "Kafka broker ${BROKER_ID} is not registered yet, retry ${i}..."
      sleep 5
    fi
  done
  echo "Failed to start Kafka broker ${BROKER_ID}." >&2
  exit 1
}

function install_and_configure_kafka_server() {
  # Find zookeeper list first, before attempting any installation.
  local zookeeper_client_port
  zookeeper_client_port=$(grep 'clientPort' /etc/zookeeper/conf/zoo.cfg |
    tail -n 1 |
    cut -d '=' -f 2)

  local zookeeper_list
  zookeeper_list=$(grep '^server\.' /etc/zookeeper/conf/zoo.cfg |
    cut -d '=' -f 2 |
    cut -d ':' -f 1 |
    sort |
    uniq |
    sed "s/$/:${zookeeper_client_port}/" |
    xargs echo |
    sed "s/ /,/g")

  if [[ -z "${zookeeper_list}" ]]; then
    # Didn't find zookeeper quorum in zoo.cfg, but possibly workers just didn't
    # bother to populate it. Check if YARN HA is configured.
    zookeeper_list=$(bdconfig get_property_value --configuration_file \
      /etc/hadoop/conf/yarn-site.xml \
      --name yarn.resourcemanager.zk-address 2>/dev/null)
  fi

  # If all attempts failed, error out.
  if [[ -z "${zookeeper_list}" ]]; then
    err 'Failed to find configured Zookeeper list; try "--num-masters=3" for HA'
  fi

  ZOOKEEPER_ADDRESS="${zookeeper_list%%,*}"

  # Install Kafka from Dataproc distro.
  install_apt_get kafka-server || dpkg -l kafka-server ||
    err 'Unable to install and find kafka-server.'

  mkdir -p /var/lib/kafka-logs
  chown kafka:kafka -R /var/lib/kafka-logs

  if [[ "${ROLE}" == "Master" ]]; then
    # For master nodes, broker ID starts from 10,000.
    if [[ "$(hostname)" == *-m ]]; then
      # non-HA
      BROKER_ID=10000
    else
      # HA
      BROKER_ID=$((10000 + $(hostname | sed 's/.*-m-\([0-9]*\)$/\1/g')))
    fi
  else
    # For worker nodes, broker ID is a random number generated less than 10000.
    # 10000 is choosen since the max broker ID allowed being set is 10000.
    BROKER_ID=$((RANDOM % 10000))
  fi
  sed -i 's|log.dirs=/tmp/kafka-logs|log.dirs=/var/lib/kafka-logs|' \
    "${KAFKA_PROP_FILE}"
  sed -i 's|^\(zookeeper\.connect=\).*|\1'${zookeeper_list}'|' \
    "${KAFKA_PROP_FILE}"
  sed -i 's,^\(broker\.id=\).*,\1'${BROKER_ID}',' \
    "${KAFKA_PROP_FILE}"
  echo -e '\nreserved.broker.max.id=100000' >>"${KAFKA_PROP_FILE}"
  echo -e '\ndelete.topic.enable=true' >>"${KAFKA_PROP_FILE}"

  if [[ "${KAFKA_ENABLE_JMX}" == "true" ]]; then
    sed -i '/kafka-run-class.sh/i export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Djava.net.preferIPv4Stack=true"' /usr/lib/kafka/bin/kafka-server-start.sh
    sed -i "/kafka-run-class.sh/i export JMX_PORT=${KAFKA_JMX_PORT}" /usr/lib/kafka/bin/kafka-server-start.sh
  fi

  wait_for_zookeeper

  # Start Kafka.
  service kafka-server restart

  wait_for_kafka
}

function install_kafka_python_package() {
  KAFKA_PYTHON_PACKAGE="kafka-python==2.0.2"
  if [[ "${INSTALL_KAFKA_PYTHON}" != "true" ]]; then
    return
  fi

  if [[ "$(echo "${DATAPROC_IMAGE_VERSION} > 2.0" | bc)" -eq 1 ]]; then
    /opt/conda/default/bin/pip install "${KAFKA_PYTHON_PACKAGE}" || { sleep 10; /opt/conda/default/bin/pip install "${KAFKA_PYTHON_PACKAGE}"; }
  else
    OS=$(. /etc/os-release && echo "${ID}")
    if [[ "${OS}" == "rocky" ]]; then
      yum install -y python2-pip
    else
      apt-get install -y python-pip
    fi
    pip2 install "${KAFKA_PYTHON_PACKAGE}" || { sleep 10; pip2 install "${KAFKA_PYTHON_PACKAGE}"; } || { sleep 10; pip install "${KAFKA_PYTHON_PACKAGE}"; }
  fi
}

function remove_old_backports {
  # This script uses 'apt-get update' and is therefore potentially dependent on
  # backports repositories which have been archived.  In order to mitigate this
  # problem, we will remove any reference to backports repos older than oldstable

  # https://github.com/GoogleCloudDataproc/initialization-actions/issues/1157
  oldstable=$(curl -s https://deb.debian.org/debian/dists/oldstable/Release | awk '/^Codename/ {print $2}');
  stable=$(curl -s https://deb.debian.org/debian/dists/stable/Release | awk '/^Codename/ {print $2}');

  matched_files="$(grep -rsil '\-backports' /etc/apt/sources.list*)"
  if [[ -n "$matched_files" ]]; then
    for filename in "$matched_files"; do
      grep -e "$oldstable-backports" -e "$stable-backports" "$filename" || \
        sed -i -e 's/^.*-backports.*$//' "$filename"
    done
  fi
}

function main() {
  OS=$(. /etc/os-release && echo "${ID}")
  if [[ ${OS} == debian ]] && [[ $(echo "${DATAPROC_IMAGE_VERSION} <= 2.1" | bc -l) == 1 ]]; then
    remove_old_backports
  fi
  recv_keys || err 'Unable to receive keys.'
  update_apt_get || err 'Unable to update packages lists.'
  install_kafka_python_package

  # Only run the installation on workers; verify zookeeper on master(s).
  if [[ "${ROLE}" == 'Master' ]]; then
    service zookeeper-server status ||
      err 'Required zookeeper-server not running on master!'
    if [[ "${RUN_ON_MASTER}" == "true" ]]; then
      # Run installation on masters.
      install_and_configure_kafka_server
    else
      # On master nodes, just install kafka command-line tools and libs but not
      # kafka-server.
      install_apt_get kafka ||
        err 'Unable to install kafka libraries on master!'
    fi
  else
    # Run installation on workers.
    install_and_configure_kafka_server
  fi
}

main

    2. Copie o script da ação de inicialização kafka.sh para o bucket do Cloud Storage. Esse script instala o Kafka em um cluster do Dataproc.

      1. 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 buckets públicos com tags regionais no Cloud Storage. Especifique uma região do Compute Engine geograficamente próxima (por exemplo, us-central1).
        • BUCKET_NAME: o nome do bucket do Cloud Storage.

    Criar um cluster do Dataproc Kafka

    1. Abra o Cloud Shell e execute o comando gcloud dataproc clusters create a seguir para criar um cluster de alta disponibilidade do Dataproc 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, números e hífens. Não pode terminar com um hífen. O nome de um cluster excluído pode ser reutilizado.
      • PROJECT_ID: o projeto a ser associado a este cluster.
      • REGION: a região do Compute Engine em que o cluster vai estar localizado, como us-central1.
        • Você pode adicionar a flag opcional --zone=ZONE para especificar uma zona dentro da região especificada, como us-central1-a. Se você não especificar uma zona, o recurso de posição de zona automática do Dataproc vai selecionar uma zona com a região especificada.
      • --image-version: a versão da imagem do Dataproc 2.1-debian11 é 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, necessário para o 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).

    Criar um tópico custdata do Kafka

    Para criar um tópico do Kafka no cluster do Kafka do Dataproc:

    1. Use o utilitário SSH para abrir uma janela de terminal na VM mestre do cluster.

    2. 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 porta 9092 no nó worker-0.

      • Você pode 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 do Kafka para gerar dados fictícios de clientes em formato CSV.

    1. 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.

    2. 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.

      Saída esperada: 

      custdata:0:10000

    Criar tabelas do Hive no Cloud Storage

    Crie tabelas do Hive para receber dados de tópicos do Kafka transmitidos. Siga as etapas abaixo para criar tabelas do Hive cust_parquet (parquet) e cust_orc (ORC) no bucket do Cloud Storage.

    1. Insira seu BUCKET_NAME no script a seguir, copie e cole o script no terminal SSH no 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 do Hive em parquet e ORC 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

    2. No terminal SSH do nó mestre do cluster do Kafka, envie o job do Hive ~/hivetables.hql para criar cust_parquet (parquet) e cust_orc (ORC) tabelas 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 já vem instalado no cluster do Dataproc Kafka. 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 Kafka custdata para tabelas do Hive

    1. Execute o comando a seguir no terminal SSH do nó mestre do cluster do Kafka para instalar a biblioteca kafka-python. Um cliente do Kafka é necessário para transmitir dados de tópicos do Kafka para o Cloud Storage. 
      pip install kafka-python

    2. Insira seu BUCKET_NAME, copie e cole o seguinte código PySpark no terminal SSH no 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 os dados para as tabelas do Hive no Cloud Storage. O formato de saída, que pode ser parquet ou ORC, é transmitido ao 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

    3. No terminal SSH do nó principal do cluster do Kafka, execute spark-submit para transmitir dados para as tabelas do Hive no Cloud Storage.

      1. Insira o nome do seu KAFKA_CLUSTER e a saída FORMAT. Em seguida, copie e cole o código abaixo no terminal SSH do nó principal do cluster do Kafka e pressione <return> para executar o código e transmitir os dados custdata do Kafka no 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 ou orc 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, especifique parquet para transmitir o tópico custdata do Kafka para a tabela parquet do Hive. Em seguida, na segunda invocação, especifique o formato orc para transmitir custdata para a tabela ORC do Hive.

    4. Depois que a saída padrão for interrompida no terminal SSH, o que significa que todos os custdata foram transmitidos, pressione <control-c> no terminal SSH para interromper o processo.

    5. Liste as tabelas do Hive no Cloud Storage.

      gcloud storage ls gs://BUCKET_NAME/tables/* --recursive

      Observações:

      • BUCKET_NAME: insira o nome do bucket do Cloud Storage que contém suas tabelas do Hive (consulte Criar tabelas do Hive).

    Consultar dados transmitidos

    1. No terminal SSH do nó mestre do cluster do Kafka, execute o seguinte comando hive para contar as mensagens custdata 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 ou cust_orc como o nome da tabela do Hive.

      Snippet de saída esperado:

    ...
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

    Excluir o projeto

    1. In the Google Cloud console, go to the Manage resources page.

      Go to Manage resources

    2. In the project list, select the project that you want to delete, and then click Delete.
    3. In the dialog, type the project ID, and then click Shut down to delete the project.

    Excluir recursos

    • In the Google Cloud console, go to the Cloud Storage Buckets page.

      Go to Buckets

    • Click the checkbox for the bucket that you want to delete.
    • To delete the bucket, click Delete, and then follow the instructions.
    • Exclua o cluster do Kafka: 
      gcloud dataproc clusters delete KAFKA_CLUSTER \
    --region=${REGION}