Apache Kafka es una plataforma de transmisión distribuida de código abierto para la integración de datos y las canalizaciones de datos en tiempo real. Proporciona un sistema de transmisión eficiente y escalable para usar en una variedad de aplicaciones, incluidas las siguientes:
- Analítica en tiempo real
- Procesamiento de transmisión
- Agregación de registros
- Mensajería distribuida
- Transmisión de eventos
Objetivos
Instalar Kafka en un clúster de HA de Dataproc con ZooKeeper (al que se hace referencia en este instructivo como “clúster de Kafka de Dataproc”).
Crear datos ficticios del cliente y, luego, publicarlos en un tema de Kafka
Crear tablas ORC y Parquet de Hive en Cloud Storage para recibir datos del tema de Kafka transmitidos
Enviar un trabajo de PySpark para suscribirse al tema de Kafka y transmitirlo en Cloud Storage en formato Parquet y ORC
Ejecutar una consulta en los datos de la tabla de Hive transmitidos para contar los mensajes de Kafka transmitidos
Costos
En este documento, usarás los siguientes componentes facturables de Google Cloud:
Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios.
Cuando finalices las tareas que se describen en este documento, puedes borrar los recursos que creaste para evitar que continúe la facturación. Para obtener más información, consulta Cómo realizar una limpieza.
Antes de comenzar
Si aún no lo hiciste, crea un proyecto de Google Cloud.
- Accede a tu cuenta de Google Cloud. Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud.
-
Habilita las API de Dataproc, Compute Engine, and Cloud Storage.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud.
-
Habilita las API de Dataproc, Compute Engine, and Cloud Storage.
- En la consola de Google Cloud, ve a la página Buckets de Cloud Storage.
- Haga clic en Crear bucket.
- En la página Crear un bucket, ingresa la información de tu bucket. Para ir al paso siguiente, haz clic en Continuar.
- En Nombre de tu bucket, ingresa un nombre que cumpla con los requisitos de nomenclatura de buckets.
-
En Elige dónde almacenar tus datos, haz lo siguiente:
- Selecciona una opción de Tipo de ubicación.
- Selecciona una opción de Ubicación.
- Para Elegir una clase de almacenamiento predeterminada para tus datos, selecciona una clase de almacenamiento.
- En Elige cómo controlar el acceso a los objetos, selecciona una opción de Control de acceso.
- Para la Configuración avanzada (opcional), especifica un método de encriptación, una política de retención o etiquetas de bucket.
- Haga clic en Crear.
Pasos del instructivo
Realiza los siguientes pasos para crear un clúster de Kafka de Dataproc y leer un tema de Kafka en Cloud Storage en formato ORC o Parquet.
Copia la secuencia de comandos de instalación de Kafka en Cloud Storage
La secuencia de comandos de la acción de inicialización kafka.sh
instala Kafka en un clúster de Dataproc.
Explora el código.
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 Dataproc.Abre Cloud Shell y ejecuta el siguiente comando:
gsutil 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 a nivel geográfico (por ejemplo:us-central1
). - BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage.
- REGION:
Crea un clúster de Kafka de Dataproc
Abre Cloud Shell y ejecuta el siguiente comando
gcloud dataproc clusters create
para crear un clúster de alta disponibilidad de Dataproc 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: El nombre del clúster, que debe ser único dentro de un proyecto. Debe comenzar con una letra minúscula y puede contener hasta 51 letras minúsculas, números y guiones. No puede terminar con un guion. El nombre de un clúster borrado se puede volver a usar.
- PROJECT_ID: 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
--zone=ZONE
opcional para especificar una zona dentro de la región especificada, comous-central1-a
. Si no especificas una zona, la función de posición de zona automática de Dataproc selecciona una zona con la región especificada.
- Puedes agregar la marca
--image-version
: Se recomienda la versión de imagen de Dataproc2.1-debian11
para este instructivo. Nota: Cada versión de la 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 Dataproc).--num-master
: Los nodos principales3
crean un clúster de alta disponibilidad. El componente Zookeeper, que requiere Kafka, está preinstalado en un clúster de HA.--enable-component-gateway
: Habilita la puerta de enlace de componentes de Dataproc.- BUCKET_NAME: Es el nombre del bucket de Cloud Storage que contiene la secuencia de comandos de inicialización
/scripts/kafka.sh
(consulta Copia la secuencia de comandos de instalación de Kafka en Cloud Storage).
Crea un tema custdata
de Kafka
Para crear un tema de Kafka en el clúster de Kafka de Dataproc, sigue estos pasos:
Usa la utilidad SSH para abrir una ventana de la terminal en la VM de la instancia principal del clúster.
Crea un tema
custdata
de 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 el agente de Kafka que se ejecuta en el puerto9092
en el nodoworker-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
En la siguiente secuencia de comandos, se usa la herramienta Kafka de kafka-console-producer.sh
para generar datos ficticios de los clientes en formato CSV.
Copia y, luego, pégala 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 del clúster de Kafka.
Ejecuta el siguiente comando de Kafka para confirmar que el tema
custdata
contenga 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 del clúster de Kafka.
Resultado esperado:
custdata:0:10000
Crea tablas de Hive en Cloud Storage
Crear tablas de Hive para recibir datos del tema de Kafka transmitidos
Realiza los siguientes pasos para crear tablas de Hive cust_parquet
(parquet) y cust_orc
(ORC) en tu bucket de Cloud Storage.
Inserta tu BUCKET_NAME en la siguiente secuencia de comandos; luego, copia y pega la secuencia de comandos en la terminal SSH del nodo instancia principal del clúster de Kafka y presiona <return> para crear una secuencia de comandos
~/hivetables.hql
(lenguaje de consulta Hive).En el siguiente paso, ejecutarás la secuencia de comandos
~/hivetables.hql
para crear tablas Parquet y ORC de Hive 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
En la terminal SSH del nodo principal de tu clúster de Kafka, envía el trabajo de Hive
~/hivetables.hql
para crear tablas de Hivecust_parquet
(parquet) ycust_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 Kafka de Dataproc. Consulta las versiones de actualización 2.1.x para obtener una lista de las versiones de los componentes de Hive incluidas en las imágenes 2.1 lanzadas recientemente.
- KAFKA_CLUSTER: Es el nombre del clúster de Kafka.
- REGION: Es la región en la que se encuentra el clúster de Kafka.
Transmite custdata
de Kafka a tablas de Hive
- Ejecuta el siguiente comando en la terminal 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 del tema de Kafka a Cloud Storage.
pip install kafka-python
Inserta tu BUCKET_NAME. Luego, copia y pega el siguiente código de PySpark en la terminal de SSH del nodo instancia principal del 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 ORC o Parquet, 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
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.Inserta el nombre de tu KAFKA_CLUSTER y el FORMAT de salida, copia y pega el siguiente código en la terminal SSH del nodo principal de tu clúster de Kafka y presiona <return> para ejecutar el código y transmitir los datos
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
oorc
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, especificaparquet
para transmitir el temacustdata
de Kafka a la tabla Parquet de Hive. Luego, en la segunda invocación, especifica el formatoorc
para transmitircustdata
a la tabla ORC de Hive.
Después de que se detenga la salida estándar en la terminal SSH, lo que significa que se transmitió todo el
custdata
, presiona <control-c> en la terminal SSH para detener el proceso.Enumerar las tablas de Hive en Cloud Storage
gsutil ls -r gs://BUCKET_NAME/tables/*
Notas:
- BUCKET_NAME: Inserta el nombre del bucket de Cloud Storage que contiene las tablas de Hive (consulta Crea tablas de Hive).
Consultar datos transmitidos
En la terminal SSH del nodo principal de tu clúster de Kafka, ejecuta el siguiente comando
hive
para contar los mensajescustdata
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
ocust_orc
como el nombre de la tabla de Hive.
Fragmento de resultado esperado:
- TABLE_NAME: Especifica
...
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)
Limpia
Borra el proyecto
Borra un proyecto de Google Cloud:
gcloud projects delete PROJECT_ID
Borrar los recursos
- Borra el depósito
gcloud storage buckets delete BUCKET_NAME
- Borra tu clúster de Kafka:
gcloud dataproc clusters delete KAFKA_CLUSTER \ --region=${REGION}