Enviar un tema de Kafka a Hive

Apache Kafka es una plataforma de streaming distribuida de código abierto para canalizaciones de datos en tiempo real e integración de datos. Proporciona un sistema de streaming eficiente y escalable para usarlo en diversas aplicaciones, entre las que se incluyen las siguientes:

  • Analíticas en tiempo real
  • Procesamiento de streaming
  • Agregación de registros
  • Mensajería distribuida
  • Streaming de eventos
.

Objetivos

  1. Instala Kafka en un clúster de alta disponibilidad de Dataproc con ZooKeeper (en este tutorial se denomina "clúster de Kafka de Dataproc").

  2. Crea datos de clientes ficticios y, a continuación, publícalos en un tema de Kafka.

  3. Crea tablas Parquet y ORC de Hive en Cloud Storage para recibir datos de temas de Kafka en streaming.

  4. Envía una tarea de PySpark para suscribirte al tema de Kafka y transmitirlo a Cloud Storage en formato Parquet y ORC.

  5. Ejecute una consulta en los datos de la tabla de Hive transmitidos para contar los mensajes de Kafka transmitidos.

Costes

En este documento, se utilizan los siguientes componentes facturables de Google Cloud:

Para generar una estimación de costes basada en el uso previsto, utiliza la calculadora de precios.

Los usuarios nuevos Google Cloud pueden disfrutar de una prueba gratuita.

Cuando termines las tareas que se describen en este documento, puedes evitar que se te siga facturando eliminando los recursos que has creado. Para obtener más información, consulta la sección Limpiar.

Antes de empezar

Si aún no lo has hecho, crea un Google Cloud proyecto.

  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.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    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.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the APIs

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

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    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.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    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. Pasos del tutorial

    Sigue estos pasos para crear un clúster de Kafka de Dataproc que lea un tema de Kafka en Cloud Storage en formato Parquet u ORC.

    Copiar la secuencia de comandos de instalación de Kafka en Cloud Storage

    El script de kafka.sh acción de inicialización instala Kafka en un clúster de Dataproc.

    1. Examina el 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. Copia la secuencia de comandos de la kafka.sh acción de inicialización en tu segmento de Cloud Storage. Esta secuencia de comandos instala Kafka en un clúster de Dataproc.

      1. Abre Cloud Shell y ejecuta el siguiente comando: 

        gcloud storage cp gs://goog-dataproc-initialization-actions-REGION/kafka/kafka.sh gs://BUCKET_NAME/scripts/

        Haz las siguientes sustituciones:

        • REGION: kafka.sh se almacena en contenedores públicos etiquetados por región en Cloud Storage. Especifica una región de Compute Engine cercana geográficamente (por ejemplo, us-central1).
        • BUCKET_NAME: nombre del segmento de Cloud Storage.

    Crear un clúster de Kafka de Dataproc

    1. Abre Cloud Shell y ejecuta el siguiente comando gcloud dataproc clusters create para crear un clúster de Dataproc de alta disponibilidad que instale los componentes Kafka y 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: el nombre del clúster, que debe ser único en un proyecto. El nombre debe empezar por una letra minúscula y puede contener hasta 51 letras minúsculas, números y guiones. No puede terminar en un guion. El nombre de un clúster eliminado se puede reutilizar.
      • PROJECT_ID: el proyecto que se va a asociar a este clúster.
      • REGION: la región de Compute Engine en la que se ubicará el clúster, como us-central1.
        • Puedes añadir la marca opcional --zone=ZONE para especificar una zona dentro de la región indicada, como us-central1-a. Si no especificas ninguna zona, la función de colocación automática de zonas de Dataproc selecciona una zona de la región especificada.
      • --image-version: En este tutorial, se recomienda usar la versión de imagen de Dataproc 2.1-debian11. Nota: Cada versión de imagen contiene un conjunto de componentes preinstalados, incluido el componente Hive que se usa en este tutorial (consulta Versiones de imagen de Dataproc compatibles).
      • --num-master: los nodos maestros 3 crean un clúster de alta disponibilidad. El componente Zookeeper, que es necesario para Kafka, está preinstalado en un clúster de alta disponibilidad.
      • --enable-component-gateway: habilita la pasarela de componentes de Dataproc.
      • BUCKET_NAME: el nombre del segmento de Cloud Storage que contiene la /scripts/kafka.sh secuencia de comandos de inicialización (consulta Copiar la secuencia de comandos de instalación de Kafka en Cloud Storage).

    Crea un custdata tema de Kafka

    Para crear un tema de Kafka en el clúster de Kafka de Dataproc, sigue estos pasos:

    1. Usa la utilidad SSH para abrir una ventana de terminal en la VM maestra del clúster.

    2. Crea un tema de custdata Kafka.

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

      Notas:

      • KAFKA_CLUSTER: inserta el nombre de tu clúster de Kafka. -w-0:9092 significa que el agente de Kafka se ejecuta en el puerto 9092 del nodo worker-0.

      • Puedes ejecutar los siguientes comandos después de crear el tema 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 contenido en el tema de Kafka custdata

    La siguiente secuencia de comandos usa la herramienta kafka-console-producer.shKafka para generar datos de clientes ficticios en formato CSV.

    1. Copia y pega el script en el terminal SSH del nodo maestro de tu clúster de Kafka. Pulsa <return> para ejecutar la secuencia de comandos.

      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: el nombre de tu clúster de Kafka.

    2. Ejecuta el siguiente comando de Kafka para confirmar que el tema custdata contiene 10.000 mensajes.

      /usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list KAFKA_CLUSTER-w-0:9092 \
--topic custdata

      Notas:

      • KAFKA_CLUSTER: el nombre de tu clúster de Kafka.

      Resultado esperado: 

      custdata:0:10000

    Crear tablas de Hive en Cloud Storage

    Crea tablas de Hive para recibir datos de temas de Kafka en streaming. Sigue estos pasos para crear tablas de Hive cust_parquet (parquet) y cust_orc (ORC) en tu cubo de Cloud Storage.

    1. Inserta tu BUCKET_NAME en la siguiente secuencia de comandos, copia y pega la secuencia de comandos en el terminal SSH del nodo maestro de tu clúster de Kafka, y pulsa <return> para crear una secuencia de comandos ~/hivetables.hql (lenguaje de consulta de Hive).

      En el siguiente paso, ejecutarás la secuencia de comandos ~/hivetables.hql para crear tablas de Hive en formato Parquet y ORC en tu bucket de 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. En el terminal SSH del nodo maestro de tu clúster de Kafka, envía el trabajo de ~/hivetables.hql Hive para crear cust_parquet (parquet) y cust_orc (ORC) tablas de Hive en tu segmento de Cloud Storage.

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

      Notas:

      • El componente Hive está preinstalado en el clúster de Kafka de Dataproc. Consulta las versiones 2.1.x para ver una lista de las versiones de componentes de Hive incluidas en las imágenes 2.1 publicadas recientemente.
      • KAFKA_CLUSTER: el nombre de tu clúster de Kafka.
      • REGION: la región en la que se encuentra tu clúster de Kafka.

    Transmitir datos de Kafka custdata a tablas de Hive

    1. Ejecuta el siguiente comando en el terminal SSH del nodo maestro de tu clúster de Kafka para instalar la biblioteca kafka-python. Se necesita un cliente de Kafka para transmitir datos de temas de Kafka a Cloud Storage. 
      pip install kafka-python

    2. Inserta tu BUCKET_NAME, copia y pega el siguiente código de PySpark en el terminal SSH del nodo maestro de tu clúster de Kafka y, a continuación, pulsa <return> para crear un archivo streamdata.py.

      La secuencia de comandos se suscribe al tema custdata de Kafka y, a continuación, transmite los datos a tus tablas de Hive en Cloud Storage. El formato de salida, que puede ser Parquet u ORC, se transfiere a la secuencia de comandos como 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. En el terminal SSH del nodo maestro de tu clúster de Kafka, ejecuta spark-submit para transmitir datos a tus tablas de Hive en Cloud Storage.

      1. Inserta el nombre de tu KAFKA_CLUSTER y la salidaFORMAT. A continuación, copia y pega el siguiente código en el terminal SSH del nodo maestro de tu clúster de Kafka y pulsa <return> para ejecutar el código y transmitir los datos de custdata de Kafka en formato Parquet a tus tablas de Hive en 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: inserta el nombre de tu clúster de Kafka.
        • FORMAT: especifica parquet o orc como formato de salida. Puedes ejecutar el comando sucesivamente para transmitir ambos formatos a las tablas de Hive. Por ejemplo, en la primera invocación, especifica parquet para transmitir el tema de Kafka custdata a la tabla Parquet de Hive. Después, en la segunda invocación, especifica el formato orc para transmitir custdata a la tabla ORC de Hive.

    4. Cuando se detenga la salida estándar en la terminal SSH, lo que significa que se ha transmitido todo el custdata, pulsa <control-c> en la terminal SSH para detener el proceso.

    5. Lista las tablas de Hive en Cloud Storage.

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

      Notas:

      • BUCKET_NAME: inserta el nombre del segmento de Cloud Storage que contiene tus tablas de Hive (consulta Crear tablas de Hive).

    Consultar datos transmitidos

    1. En el terminal SSH del nodo maestro de tu clúster de Kafka, ejecuta el siguiente comando hive para contar los mensajes de Kafka custdata transmitidos en las tablas de Hive de Cloud Storage.

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

      Notas:

      • TABLE_NAME: especifica cust_parquet o cust_orc como nombre de tabla de Hive.

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

    Limpieza

    Eliminar el proyecto

    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.

    Eliminar 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.
    • Elimina tu clúster de Kafka: 
      gcloud dataproc clusters delete KAFKA_CLUSTER \
    --region=${REGION}