Apache Kafka es una plataforma de transmisión distribuida de código abierto para canalizaciones de datos en tiempo real y la integración de datos. 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
Pasos del instructivo
Sigue estos pasos para crear un clúster de Kafka de Dataproc para 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 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, 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.
- REGION:
Crea un clúster de Dataproc Kafka
Abre Cloud Shell y, luego, ejecuta el siguiente comando
gcloud dataproc clusters create
para crear un clúster de Dataproc de alta disponibilidad 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, comous-central1-a
. Si no especificas una zona, la función de ubicación de zona automática de Dataproc selecciona una zona con la región especificada.
- Puedes agregar la marca opcional
--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 las versiones de imágenes de Dataproc compatibles).--num-master
: Los nodos principales de3
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 de Dataproc.- 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 Dataproc, haz lo siguiente:
Usa la utilidad SSH para abrir una ventana de terminal en la VM instancia principal del clúster.
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 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
La siguiente secuencia de comandos usa la herramienta kafka-console-producer.sh
de Kafka para generar datos de clientes ficticios en formato CSV.
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.
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.
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
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 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 Dataproc Kafka. 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
- 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
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
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 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
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 de Hive en formato Parquet. Luego, en la segunda invocación, especifica el formatoorc
para transmitircustdata
a la tabla de Hive en formato ORC.
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.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
En la terminal de 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 del 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)