Transmite un tema de Kafka a Hive

Pasos del instructivo

Sigue estos pasos para crear un clúster de Managed Service para Apache Spark Kafka y leer un tema de Kafka en Cloud Storage en formato Parquet O ORC.

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

La secuencia de comandos de la kafka.sh acción de inicialización instala Kafka en un clúster de Managed Service para Apache Spark.

1. Explora el código.

   Expandirme Contraerme

   #!/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() {
     if [[ ${OS} == debian ]] && [[ $(echo "${DATAPROC_IMAGE_VERSION} >= 3.0" | bc -l) == 1 ]]; then
       retry_apt_command "apt-get update && apt-get install -y gnupg"
       export GNUPGHOME="$(mktemp -d)"
       trap 'rm -rf "${GNUPGHOME}"' EXIT
       gpg --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C
       mkdir -p /etc/apt/trusted.gpg.d
       gpg --export B7B3B788A8D3785C > /etc/apt/trusted.gpg.d/mysql-repo.gpg
     else
       retry_apt_command "apt-get install -y gnupg2 && \
         apt-key adv --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C"
     fi
   }
   
   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 acción de inicialización kafka.sh en tu bucket de Cloud Storage. Con esta secuencia de comandos, se instala Kafka en un clúster de Managed Service para Apache Spark.

   1. Abre Cloud Shell y, luego, ejecuta el siguiente comando:

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

      Realiza los siguientes reemplazos:

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

Crea un clúster de Managed Service para Apache Spark Kafka

1. Abre Cloud Shell y, luego, ejecuta el siguiente comando de gcloud dataproc clusters create para crear un clúster de clúster de AA de Managed Service para Apache Spark que instale los componentes de 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: Es el nombre del clúster, que debe ser único dentro de un proyecto. El nombre debe comenzar con una letra en minúscula y puede contener hasta 51 letras en minúscula, números y guiones. No puede terminar con un guion. Se puede reutilizar el nombre de un clúster borrado.
   • PROJECT_ID: Es el proyecto que se asociará con este clúster.
   • REGION: Es la región de Compute Engine en la que se ubicará el clúster, como us-central1.
     • Puedes agregar la marca opcional --zone=ZONE para especificar una zona dentro de la región especificada, como us-central1-a. Si no especificas una zona, la función de colocación automática de zonas de Managed Service for Apache Spark selecciona una zona con la región especificada.
   • --image-version: Se recomienda la versión de imagen del servicio administrado para Apache Spark 2.1-debian11 para este instructivo. Nota: Cada versión de imagen contiene un conjunto de componentes preinstalados, incluido el componente de Hive que se usa en este instructivo (consulta Versiones de imágenes compatibles de Managed Service for Apache Spark).
   • --num-master: Los nodos principales de 3 crean un clúster de HA. El componente Zookeeper, que requiere Kafka, está preinstalado en un clúster de alta disponibilidad.
   • --enable-component-gateway: Habilita la puerta de enlace de componentes del servicio administrado para Apache Spark.
   • BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage que contiene el script de inicialización de /scripts/kafka.sh (consulta Copia el script de instalación de Kafka en Cloud Storage).

Crea un tema de Kafka custdata

Para crear un tema de Kafka en el clúster de Kafka de Managed Service para Apache Spark, haz lo siguiente:

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

2. Crea un tema de Kafka custdata.

   /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 el agente de Kafka que se ejecuta en el puerto 9092 en el 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
     

Publica contenido en el tema custdata de Kafka

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

1. Copia y pega la secuencia de comandos en la terminal SSH del nodo principal de tu clúster de Kafka. Presiona <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: Es 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: Es el nombre de tu clúster de Kafka.

   Resultado esperado:

   custdata:0:10000
   

Crea tablas de Hive en Cloud Storage

Crea tablas de Hive para recibir los datos del tema de Kafka transmitidos. Sigue estos pasos para crear tablas de Hive cust_parquet (Parquet) y cust_orc (ORC) en tu bucket de Cloud Storage.

1. Inserta tu BUCKET_NAME en la siguiente secuencia de comandos, luego, cópiala y pégala en la terminal de SSH del nodo principal de tu clúster de Kafka y, luego, presiona <return> para crear una secuencia de comandos de ~/hivetables.hql (lenguaje de consultas de Hive).

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

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

   Notas:

   • El componente de Hive está preinstalado en el clúster de Managed Service para Apache Kafka Spark. Consulta las versiones de actualización 2.1.x para obtener una lista de las versiones del componente Hive incluidas en las imágenes 2.1 lanzadas recientemente.
   • KAFKA_CLUSTER: Es el nombre de tu clúster de Kafka.
   • REGION: Es la región en la que se encuentra tu clúster de Kafka.

Transmite Kafka custdata a tablas de Hive

1. Ejecuta el siguiente comando en la terminal de SSH del nodo principal 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, luego copia y pega el siguiente código de PySpark en la terminal SSH del nodo principal de tu clúster de Kafka y, luego, presiona <return> para crear un archivo streamdata.py.

   La secuencia de comandos se suscribe al tema custdata de Kafka y, luego, transmite los datos a tus tablas de Hive en Cloud Storage. El formato de salida, que puede ser Parquet o ORC, se pasa a la secuencia de comandos como un 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 la terminal SSH del nodo principal 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 salida FORMAT. Luego, copia y pega el siguiente código en la terminal SSH del nodo principal de tu clúster de Kafka y, luego, presiona <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 el formato de salida. Puedes ejecutar el comando de forma sucesiva para transmitir ambos formatos a las tablas de Hive. Por ejemplo, en la primera invocación, especifica parquet para transmitir el tema custdata de Kafka a la tabla de Hive en formato Parquet. Luego, en la segunda invocación, especifica el formato orc para transmitir custdata a la tabla de Hive en formato ORC.

4. Después de que se detenga la salida estándar en la terminal SSH, lo que indica que se transmitió todo el contenido de custdata, presiona <control-c> en la terminal SSH para detener el proceso.

5. Enumera las tablas de Hive en Cloud Storage.

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

   Notas:

   • BUCKET_NAME: Inserta el nombre del bucket de Cloud Storage que contiene tus tablas de Hive (consulta Cómo crear tablas de Hive).

Consulta datos transmitidos

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

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

   Notas:

   • TABLE_NAME: Especifica cust_parquet o cust_orc como el nombre de la tabla de Hive.

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