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
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.
Examina el código.
Copia la secuencia de comandos de la
kafka.shacción de inicialización en tu segmento de Cloud Storage. Esta secuencia de comandos instala Kafka en un clúster de Dataproc.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.shse 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.
- REGION:
Crear un clúster de Kafka de Dataproc
Abre Cloud Shell y ejecuta el siguiente comando
gcloud dataproc clusters createpara 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. --image-version: En este tutorial, se recomienda usar la versión de imagen de Dataproc2.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 maestros3crean 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 secuencia de comandos de inicialización de
/scripts/kafka.sh(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:
Usa la utilidad SSH para abrir una ventana de terminal en la VM maestra del clúster.
Crea un tema de
custdataKafka./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:9092significa que el agente de Kafka se ejecuta en el puerto9092del 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
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.
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.
Ejecuta el siguiente comando de Kafka para confirmar que el tema
custdatacontiene 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.
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.hqlpara 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 el terminal SSH del nodo maestro de tu clúster de Kafka, envía el trabajo de
~/hivetables.hqlHive para crearcust_parquet(parquet) y una tabla de Hivecust_orc(ORC) 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
- 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
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
custdatade 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]) EOFEn el terminal SSH del nodo maestro de tu clúster de Kafka, ejecuta
spark-submitpara transmitir datos a tus tablas de Hive en Cloud Storage.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
custdatade 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 FORMATNotas:
- KAFKA_CLUSTER: inserta el nombre de tu clúster de Kafka.
- FORMAT: especifica
parquetoorccomo formato de salida. Puedes ejecutar el comando sucesivamente para transmitir ambos formatos a las tablas de Hive. Por ejemplo, en la primera invocación, especificaparquetpara transmitir el tema de Kafkacustdataa la tabla Parquet de Hive. Después, en la segunda invocación, especifica el formatoorcpara transmitircustdataa la tabla ORC de Hive.
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.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
En el terminal SSH del nodo maestro de tu clúster de Kafka, ejecuta el siguiente comando
hivepara contar los mensajes de Kafkacustdatatransmitidos en las tablas de Hive de Cloud Storage.hive -e "select count(1) from TABLE_NAME"
Notas:
- TABLE_NAME: especifica
cust_parquetocust_orccomo nombre de tabla de Hive.
Fragmento de salida 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)