Faça stream de 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. Oferece um sistema de streaming eficiente e escalável para utilização numa variedade de aplicações, incluindo:

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

Passos do tutorial

Siga os passos abaixo para criar um cluster do Dataproc Kafka para ler um tópico do Kafka no Cloud Storage no formato parquet OU ORC.

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

O script de kafka.sh ação de inicialização instala o Kafka num 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 kafka.sh ação de inicialização para o seu contentor do Cloud Storage. Este script instala o Kafka num cluster do Dataproc.

    1. Abra o Cloud Shell e, de seguida, 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 contentores públicos etiquetados regionalmente no Cloud Storage. Especifique uma região do Compute Engine geograficamente próxima (exemplo: us-central1).
      • BUCKET_NAME: o nome do seu contentor do Cloud Storage.

Crie um cluster do Dataproc Kafka

  1. Abra o Cloud Shell e, de seguida, execute o seguinte comando gcloud dataproc clusters create para criar um cluster HA 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

    Notas:

    • KAFKA_CLUSTER: o nome do cluster, que tem de ser exclusivo num projeto. O nome tem de começar por uma letra minúscula e pode conter até 51 letras minúsculas, números e hífenes. Não pode terminar com um hífen. O nome de um cluster eliminado pode ser reutilizado.
    • PROJECT_ID: o projeto a associar a este cluster.
    • REGION: a região do Compute Engine onde o cluster vai estar localizado, como us-central1.
      • Pode adicionar a flag --zone=ZONE opcional para especificar uma zona na região especificada, como us-central1-a. Se não especificar uma zona, a funcionalidade de posicionamento automático de zonas do Dataproc seleciona uma zona com a região especificada.
    • --image-version: a versão da imagem do Dataproc 2.1-debian11 é recomendada para este tutorial. Nota: cada versão da imagem contém um conjunto de componentes pré-instalados, incluindo o componente Hive usado neste tutorial (consulte Versões de imagens do Dataproc suportadas).
    • --num-master: os nós principais 3 criam um cluster de HA. O componente Zookeeper, necessário para o Kafka, está pré-instalado num cluster de alta disponibilidade.
    • --enable-component-gateway: ativa o gateway de componentes do Dataproc.
    • BUCKET_NAME: O nome do seu contentor do Cloud Storage que contém o /scripts/kafka.shscript de inicialização (consulte Copie o script de instalação do Kafka para o Cloud Storage).

Crie um tópico do Kafka custdata

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 principal do cluster.

  2. Crie um tópico do Kafka custdata.

    /usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --create --topic custdata

    Notas:

    • KAFKA_CLUSTER: insira o nome do seu cluster Kafka. -w-0:9092 significa o agente Kafka em execução na porta 9092 no nó worker-0.

    • 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 do Kafka custdata

O script seguinte usa a ferramenta kafka-console-producer.shKafka para gerar dados de clientes fictícios no formato CSV.

  1. Copie e, de seguida, cole o script no terminal SSH no nó principal do cluster Kafka. Prima <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=:"

    Notas:

    • KAFKA_CLUSTER: o nome do seu cluster Kafka.

  2. Execute o seguinte comando do Kafka para confirmar que 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

    Notas:

    • KAFKA_CLUSTER: o nome do seu cluster Kafka.

    Resultado esperado: 

    custdata:0:10000

Crie tabelas Hive no Cloud Storage

Crie tabelas Hive para receber dados de tópicos Kafka transmitidos. Execute os seguintes passos para criar cust_parquet (parquet) e cust_orc (ORC) tabelas Hive no seu contentor do Cloud Storage.

  1. Insira o seu BUCKET_NAME no seguinte script, em seguida, copie e cole o script no terminal SSH no nó principal do cluster Kafka, depois, prima <return> para criar um script ~/hivetables.hql (Hive Query Language).

    No passo seguinte, vai executar o script ~/hivetables.hql para criar tabelas Hive parquet e ORC no seu contentor 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 no nó principal do cluster Kafka, envie a tarefa do Hive para criar cust_parquet (parquet) e tabelas do Hive cust_orc (ORC) no seu contentor do Cloud Storage.~/hivetables.hql

    gcloud dataproc jobs submit hive \
    --cluster=KAFKA_CLUSTER \
    --region=REGION \
    -f ~/hivetables.hql

    Notas:

    • O componente Hive está pré-instalado no cluster do Dataproc Kafka. Consulte as versões de lançamento 2.1.x para ver uma lista das versões dos componentes do Hive incluídas nas imagens 2.1 lançadas recentemente.
    • KAFKA_CLUSTER: o nome do seu cluster Kafka.
    • REGION: a região onde o cluster Kafka está localizado.

Faça stream de Kafka custdata para tabelas do Hive

  1. Execute o seguinte comando no terminal SSH no nó principal do cluster Kafka para instalar a biblioteca kafka-python. É necessário um cliente Kafka para fazer stream de dados de tópicos do Kafka para o Cloud Storage. 
    pip install kafka-python

  2. Insira o seu BUCKET_NAME, copie e, em seguida, cole o seguinte código PySpark no terminal SSH no nó principal do cluster Kafka e, de seguida, prima <return> para criar um ficheiro streamdata.py.

    O script subscreve o tópico do Kafka custdata e, em seguida, transmite os dados para as suas 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

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

    1. Insira o nome do seu KAFKA_CLUSTER e a saída FORMAT. Em seguida, copie e cole o seguinte código no terminal SSH no nó principal do cluster Kafka e, de seguida, prima <return> para executar o código e transmitir os dados custdata do Kafka no formato Parquet para as suas tabelas 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

      Notas:

      • KAFKA_CLUSTER: insira o nome do seu cluster Kafka.
      • FORMAT: especifique parquet ou orc como o formato de saída. Pode executar o comando sucessivamente para fazer stream de ambos os formatos para as tabelas do Hive. Por exemplo, na primeira invocação, especifique parquet para fazer stream do tópico do Kafka custdata para a tabela parquet do Hive. Em seguida, na segunda invocação, especifique o formato orc para fazer stream de custdata para a tabela ORC do Hive.

  4. Depois de a saída padrão parar no terminal SSH, o que significa que todo o custdata foi transmitido, prima <control-c> no terminal SSH para parar o processo.

  5. Liste as tabelas do Hive no Cloud Storage.

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

    Notas:

    • BUCKET_NAME: Insira o nome do contentor do Cloud Storage que contém as suas tabelas do Hive (consulte o artigo Crie tabelas do Hive).

Consulte dados transmitidos em streaming

  1. No terminal SSH no nó principal do cluster Kafka, execute o seguinte comando hive para contar as mensagens custdata do Kafka transmitidas nas tabelas Hive no Cloud Storage.

    hive -e "select count(1) from TABLE_NAME"

    Notas:

    • TABLE_NAME: especifique cust_parquet ou cust_orc como o nome da tabela do Hive.

    Fragmento 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)