Nota:

Transmisión de mensajes AVRO mediante Oracle Cloud Infrastructure Streaming y OCI Data Flow con procesamiento de microbloques

Introducción

En el panorama actual basado en datos, la capacidad de procesar y analizar flujos de datos en tiempo real es crucial para las empresas que buscan obtener información y responder rápidamente a las condiciones cambiantes. Las tecnologías de procesamiento de datos de transmisión han surgido como soluciones potentes para el manejo de flujos de datos continuos de gran volumen. En este tutorial, exploraremos un enfoque innovador para transmitir de forma eficaz mensajes AVRO mediante Oracle Cloud Infrastructure (OCI) Streaming, junto con la técnica de procesamiento de micro lotes y mejorado con las capacidades sin servidor de Oracle Functions basadas en el proyecto FN de código abierto.

Introducción a AVRO y datos de flujo

AVRO, un formato de serialización de datos ampliamente adoptado, es conocido por su eficacia en la representación de estructuras de datos complejas y su compatibilidad con varios lenguajes de programación. Cuando se integra con tecnologías de transmisión, AVRO permite a las organizaciones transmitir y procesar datos casi en tiempo real, lo que les permite extraer valiosas estadísticas sin la latencia que normalmente se asocia al procesamiento por lotes.

OCI Streaming: potencia los datos en tiempo real

Oracle Cloud Infrastructure (OCI) tiene un conjunto de herramientas para manejar datos en la nube, siendo OCI Streaming uno de estos servicios personalizados para flujos de datos de alto rendimiento en tiempo real. Al aprovechar OCI Streaming, los desarrolladores pueden construir pipelines de datos escalables y fiables que ingieran, procesen y distribuyan flujos de datos de forma eficiente.

Spark gestionado por OCI Data Flow: una solución sin dependencia

Oracle Cloud Infrastructure (OCI) Data Flow es un servicio Apache Spark totalmente gestionado que realiza tareas de procesamiento en conjuntos de datos de gran tamaño sin que la infraestructura se despliegue o gestione.

Procesamiento de microlotes El procesamiento de microlotes implica desglosar los flujos de datos entrantes en lotes compactos utilizando el tiempo o el tamaño como criterios. Estos lotes se procesan como trabajos más pequeños. A diferencia del manejo constante e individual de los registros en el procesamiento de flujos, el procesamiento de micro lotes introduce un poco de retraso y almacenamiento antes del procesamiento, lo que permite un mayor control sobre qué hacer con los datos. A diferencia del procesamiento por lotes tradicional que aborda los conjuntos de big data a intervalos, el procesamiento de micro lotes proporciona un procesamiento y una entrega de resultados casi en tiempo real.

Desbloquear la sinergia: OCI Streaming, OCI Data Flow y Oracle Functions

En este tutorial se explica la fusión de OCI Streaming, OCI Data Flow gestionó Spark Streaming y Oracle Functions. Le guiaremos a través del proceso de configuración de un pipeline de datos de transmisión de extremo a extremo que ingiere mensajes codificados por AVRO, los procesa de forma eficaz mediante las capacidades de procesamiento de microlotes de Spark gestionado por OCI Data Flow e introduce el procesamiento sin servidor basado en eventos con Oracle Functions.

Objetivo

Utilice el procesamiento de microlotes de Spark gestionado por OCI Streaming y OCI Data Flow para crear un pipeline de procesamiento de datos eficiente en tiempo real con formato AVRO.

IMPORTANTE: este tutorial está diseñado exclusivamente para fines educativos y de estudio. Proporciona un entorno para que los alumnos experimenten y obtengan experiencia práctica en un entorno controlado. Es crucial tener en cuenta que las configuraciones y prácticas de seguridad empleadas en este laboratorio podrían no ser adecuadas para escenarios del mundo real.

Las consideraciones de seguridad para las aplicaciones reales suelen ser mucho más complejas y dinámicas. Por lo tanto, antes de implementar cualquiera de las técnicas o configuraciones demostradas aquí en un entorno de producción, es esencial realizar una evaluación y revisión exhaustivas de la seguridad. Esta revisión debe abarcar todos los aspectos de la seguridad, incluidos el control de acceso, el cifrado, la supervisión y el cumplimiento, para garantizar que el sistema se ajuste a las políticas y los estándares de seguridad de la organización.

La seguridad siempre debe ser una prioridad máxima al realizar la transición de un entorno de prácticas a un despliegue real.

Flujo de Proceso
T0_1

Arquitectura de alto nivel
T0_1

Requisitos - Oracle Cloud Infrastructure

Requisitos previos - Entorno de máquina local

Tarea 1: Configuración de grupos dinámicos

  1. Vaya al dominio, haga clic en Grupos dinámicos y cree los siguientes grupos.

    Nombre del grupo: MyFunctions

    ALL {resource.type = 'fnfunc', resource.compartment.id = 'pasteYourCompartmentID'}

    Nombre del grupo: ContainerIntances

    ALL {resource.type='compute-container-instances',  resource.compartment.id = 'pasteYourCompartmentID'}

    Nombre del grupo: DataFlowDynamicGroup

    ALL {resource.type='dataflowrun', resource.compartment.id = 'pasteYourCompartmentID'}

Tarea 2: Crear políticas

Tarea 3: Crear cubos de almacenamiento y cargar esquema AVRO

  1. Vaya a Cubos, haga clic en Crear cubo y cree un nuevo cubo denominado avro-schema-bucket para almacenar el archivo de esquema de AVRO.

    T3_1

  2. Ahora, seleccione su cubo y TAKE NOTE of the Namespace, lo necesitaremos más tarde.

    T3_1

  3. Cargue el archivo user.asvc en este cubo creado.

    T3_1

Tarea 4: Creación del tema de flujo de OCI privado

  1. Vaya a Análisis e IA y, a continuación, haga clic en Flujo y cree un nuevo flujo denominado FrontDoorTopic.

    T4_0

  2. Seleccione Pools de flujos, haga clic en PrivatePool y, a continuación, haga clic en la opción Configuración de conexión de Kafka y en TAREA NOTA de los campos. Lo necesitaremos más tarde.

    T4_0

Tarea 5: Creación del TOKEN AUTH

Cree el TOKEN AUTH para su usuario; es necesario para trabajar con el tema de Kafka

  1. Haga clic en el ícono de usuario superior derecho y, luego, seleccione la opción User Settings (Configuración de usuario).

  2. Haga clic en Tokens de autenticación y, a continuación, genere un nuevo token y TAKE NOTE de su token.

    T4_1

Tarea 6: Crear registro de contenedor

  1. Vaya al menú Servicios para desarrolladores, haga clic en Registro de contenedor y cree los siguientes repositorios privados.

    Nombre del Repositorio Tipo
    api-avro-sample_a Privadas
    api-avro-sample_b Privadas
    fn-recep-avro Privadas

  2. Compruebe los repositorios y TAKE NOTE del espacio de nombres.

    T6_1

  3. Abra el shell de terminal en el que tiene instalado OCI CLI y Docker, y continúe con la conexión en el registro. Compruebe cuál es la URL correcta para su REGION. En este tutorial, estamos utilizando Este de Brasil (São Paulo), donde la URL del registro es gru.ocir.io.

    docker login gru.ocir.io
Username: <your container namespace>/youruser
Password: YOUR_AUTH_TOKEN_CREATED_EARLIER

    T6_1

Tarea 7: Creación de OCI Vault

Cree OCI Vault y proporcione las variables necesarias que se utilizarán más adelante en este tutorial.

  1. Vaya a Identificar y seguridad, haga clic en Almacén y, a continuación, haga clic en Crear almacén.

    T7_1new

  2. Seleccione el nuevo almacén y cree Claves de cifrado maestras para él.

    T7_1new

  3. Cree un nuevo secreto denominado AUTH_KEY y pegue la clave de autenticación que ha creado anteriormente.

    T7_1new

  4. Repita el proceso de creación de secretos y cree los siguientes secretos nuevos:

    Nombre Variable Valor
    KAFKA_BOOTSTRAPSERVER "El servidor pushrap de la configuración de OCI Streaming"
    KAFKA_TOPIC “FrontDoorTopic”
    KAFKA_USERNAME "Su nombre de usuario + ID de pool de flujos de la configuración de OCI Streaming"
    AUTH_KEY "Su token AUTH que ha creado en pasos anteriores"

  5. Tome nota del OCID secreto que se creó para cada secreto y cree un nuevo archivo de configuración.

    • El archivo config.properties contiene las variables de asignación de la aplicación a los ocids secretos de almacén. La aplicación utilizará este archivo para identificar los secretos de almacén que se deben recopilar durante el tiempo de ejecución.

    • Cree un nuevo archivo en la máquina local en la que tenga acceso a OCI-CLI:
      Sustituya por su OCID por cada Secreat
      Nombre de archivo: config.properties

      kafka_bootstrapserver_vaultOCID=ocid1.vaultsecret.REPLACE-WITH-YOURS
kafka_topic_vaultOCID=ocid1.vaultsecret.REPLACE-WITH-YOURSxxxxxx
kafka_username_vaultOCID=ocid1.vaultsecret.REPLACE-WITH-YOURS
auth_token_vaultOCID=ocid1.vaultsecret.oc1.REPLACE-WITH-YOURS

  6. Vaya a Cubos, haga clic en Crear cubo y cree un nuevo cubo denominado config para almacenar el archivo config.properties.

  7. Cargue el archivo config.properties en el cubo de almacenamiento config

    ls -lrt config.properties
oci os object put -bn config --file config.properties --force

    T7_1new

Tarea 8: Crear un mensaje AVRO simple y guardarlo en un archivo utilizando el código de python de ejemplo proporcionado

  1. Abra el terminal de shell donde tenga instalado OCI CLI, Docker y Python3 y cree un nuevo archivo AVRO que contenga un único mensaje basado en el mismo esquema de AVRO que cargamos anteriormente en Object Storage.

    Nota: Debe comprobar la versión de Python; estamos utilizando Python 3.9.16 en este tutorial; es posible que las versiones anteriores no funcionen.

  2. Consiga el código desde aquí Create_avro_sample.zip.

  3. Descomprímalo en la ubicación que elija y, a continuación, ejecute el programa para generar un mensaje AVRO de ejemplo:

    cd ~
mkdir create_avro_sample
cd create_avro_sample
unzip CreateAVRO_SampleFile.zip
# Check the files are there
ls -lrt
# install the python dependencies on requirements.txt
pip3 install -r requirements.txt
# Run the program and create an AVRO message file
python3 create_avro_sample.py meu_file.bin '{"id":10029,"name":"John","email":"john@bla.com"}'

    T8_1 T8_1 T8_1

Tarea 9: Creación de la función de OCI para recibir el mensaje AVRO y publicar en el tema OCI Streaming

  1. Vaya a Servicios para desarrolladores, en Funciones, haga clic en Aplicaciones y, a continuación, en Crear aplicación.

    T9_1

  2. Vaya al shell de terminal donde tiene Docker, OCI CLI, FN CLI instalado y ejecute los siguientes comandos para inicializar la función.

    Nota: Si ha seguido los pasos, el comando de conexión de Docker ya se ha ejecutado, si no, continúe con el paso de conexión de Docker en la tarea Crear registro de contenedor.

    fn create context oci-cloud --provider oracle
fn use context oci-cloud
fn update context oracle.compartment-id PASTE_YOUR_COMPARTMENT_OCID
fn update context api-url https://functions.sa-saopaulo-1.oraclecloud.com
fn update context registry gru.ocir.io/PASTE_YOUR_REGISTRY_NAMESPACE

    Nota: En este tutorial, estamos utilizando la región Este de Brasil (São Paulo), si está utilizando una región diferente, debe cambiar las ubicaciones api-url y registry.

    T9_1

  3. Cree una función Hello-world simple para asegurarse de que todos los valores son correctos.

    fn init --runtime python fn-recep-avro
cd fn-recep-avro
fn deploy --app MyReceptionApp
fn invoke MyReceptionApp fn-recep-avro

    T9_1

  4. Obtenga el código de ejemplo de la función AVRO en el archivo fn-recep-avro.zip y reemplace el código hello-world que creamos anteriormente. Debe obtener ambos archivos func.py y requirements.txt para poder trabajar.

    # Check you have the right code for func.py & requirements.txt (you got from zip file)
ls -lrt

    T9_1

  5. Crear el nuevo código y desplegar la función

    fn deploy --app MyReceptionApp

    T9_1

  6. Para llamar a la función, debemos transferir un mensaje AVRO como parámetro, para ello, utilizaremos el archivo de mensaje AVRO de ejemplo creado a partir de los pasos anteriores. La primera vez que se llama a una función tarda un poco más desde que se debe iniciar.

    # Check where you created the sample avro message file
ls -lrt ../create_avro_sample/

# Invoke the function to check if it's working as expected
echo -n -e "$(cat ../create_avro_sample/meu_file.bin)" | fn invoke MyReceptionApp fn-recep-avro

    T9_1

Tarea 10: Crear gateway de API para exponer la función

  1. En la consola, haga clic en Servicios de desarrollador y, a continuación, en Gestión de API, haga clic en Gateways y, a continuación, en Crear gateway.

    T10_1

  2. Una vez creado, haga clic en la opción Despliegues y, a continuación, haga clic en Crear despliegue.

    Nombre: RecepFunction
    Prefijo de ruta de acceso: /

    • En Autenticación, seleccione Sin autenticación, ya que se trata de una práctica sencilla y no se ha implantado ninguna autenticación de API. El objetivo principal aquí es demostrar una llamada HTTPS que transfiere un mensaje AVRO binario a través de la API y, para el propósito de este laboratorio, no implantaremos ningún método de autenticación para este laboratorio simple.
    • Antes de seguir adelante con un entorno real, asegúrese de seguir las mejores prácticas de seguridad para el gateway de API.
    • Para obtener más información, consulte Protección de gateways y recursos de API.

    Ruta 1: Ruta: /

    Metas: POST
    Tipo de backend: funciones de Oracle
    Aplicación: seleccione la función

    T9_1

    T9_1

    T9_1

    T9_1

  3. Compruebe el punto final del gateway de API y tome nota de ello.

    T9_1

  4. Abra el terminal de shell de Linux y llame a API Gateway. Sustituya la URL de API por el punto final correcto que ha obtenido en el paso anterior.

    cd create_avro_sample/
ls -lrt
curl -X POST -H "Content-Type: application/octet-stream" \
     -d "$(echo -n -e "$(cat meu_file.bin)")" \
     https://xxxxxxxxxxxxx.apigateway.sa-saopaulo-1.oci.customer-oci.com/

    T9_1

Punto de Control

T9_1

Tarea 11: Crear imagen de contenedor para el tipo de API A

Nota: Las API de tipo A y código B son básicamente las mismas con un solo mensaje de cabecera diferente para simular dos API diferentes.

  1. Obtenga el código de la API type A y descomprímalo en el terminal de shell de Linux api-avro-sample_a.zip.

  2. Obtenga el espacio de nombres del registro de contenedor que ha obtenido en los pasos anteriores y cree la ubicación del registro de aplicación siguiendo el patrón siguiente. ocir url se basa en la región, es decir, gru.ocir.io para Brasil East(SaoPaulo)

    [ocir url]/[su espacio de nombres]/api-avro-sample_a:latest

  3. En el terminal de shell de Linux, cree y transfiera la imagen de docker para esta API.

    ls -lrt
docker build . -t gru.ocir.io/yournamespace/api-avro-sample_a:latest
docker push gru.ocir.io/yournamespace/api-avro-sample_a:latest

    T10_1 T10_1

Tarea 12: Crear imagen de contenedor para el tipo de API B

  1. Obtenga el código de la API type B y descomprímalo en el terminal de shell de Linux api-avro-sample_b.zip.

  2. Obtenga el espacio de nombres del registro de contenedor que ha obtenido en los pasos anteriores y cree la ubicación del registro de aplicación siguiendo el patrón siguiente. ocir url se basa en la región, es decir, gru.ocir.io para Brasil East(SaoPaulo)

    [ocir url]/[su espacio de nombres]/api-avro-sample_b:latest

  3. En el terminal de shell de Linux, cree y transfiera la imagen de docker para esta API.

    ls -lrt
docker build . -t gru.ocir.io/yournamespace/api-avro-sample_b:latest
docker push gru.ocir.io/yournamespace/api-avro-sample_b:latest

    T10_1

    T10_1

  4. Compruebe en la página de registro del contenedor si la imagen se ha transferido correctamente.

    T10_1

Tarea 13: Despliegue de las API en el servicio de contenedor

  1. Vaya a Servicios para desarrolladores, Instancias de contenedor y haga clic en Crear instancia de contenedor.

    T13_1

    T13_1

  2. Repita el paso 1 para api-type-b y seleccione la imagen correcta para la API de TYPE B.

    1. Vaya a Servicios para desarrolladores, Instancias de contenedor, haga clic en Crear instancia de contenedor y repita los pasos para desplegar el tipo de API B

    2. Obtenga la dirección FQDN interna de las instancias de contenedor.

      T14_1

      • Haga clic en la instancia de contenedor y tome nota de cada dirección FQDN interna.

      T14_1

    3. Vaya a Identificar y ejecutar, haga clic en Almacén, seleccione su almacén y cree dos secretos nuevos.

      Nombre secreto Valor
      API_TYPE_A_URL pegar la dirección privada FQDN interna para el tipo de API A
      API_TYPE_B_URL pegar la dirección privada FQDN interna para el tipo de API B

      Tome nota de cada OCID secreto

      Su bóveda debe quedar así ahora:

      T14_1

    4. Edite el archivo config.properties que ha cargado en el cubo de almacenamiento config y agregue nuevas entradas para el OCID secreto

      ls -lrt config.properties
vi config.properties
api_type_a_url_vaultOCID=paste_API_TYPE_A_URL_secretOCID
api_type_b_url_vaultOCID=paste_API_TYPE_B_URL_secretOCID

# After save, please upload the new version to Object Storage
cat config.properties
oci os object put -bn config --file config.properties --force

      El archivo se debe parecer al siguiente:
      T14_1

      T14_1

Tarea 14: prueba de las API mediante create_avro_sample.py

  1. Vaya al terminal del shell de Linux donde guardó create_avro_sample.py en la Tarea 7 y cree nuevos mensajes para probar las llamadas de API. Estamos creando dos nuevos archivos AVRO con diferentes ID (1010 y 1020) que usaremos como filtro dentro del programa Spark Stream (DataFlow).

    ls -lrt
python3 create_avro_sample.py type_a_message.bin '{"id":1010,"name":"Paul that goes to API type A","email":"paul@bla.com"}'

python3 create_avro_sample.py type_b_message.bin '{"id":1020,"name":"Mary that goes to API type B","email":"mary@bla.com"}'

    T14_1

  2. La llamada a las API que transfieren el mensaje AVRO para probar funciona correctamente. Vaya a la página Instancias de contenedor y obtenga la dirección FQDN interna para cada una de las API api-type-a y api-type-b. Recuerde reemplazar la siguiente URL para la dirección FQDN interna correspondiente de sus API.

    ls -lrt type*

curl -i -X POST -H "Content-Type: application/octet-stream" \
   --data-binary "@type_a_message.bin" \
   xxx.xx.x.xxx

curl -i -X POST -H "Content-Type: application/octet-stream" \
   --data-binary "@type_b_message.bin" \
   xxx.xxx.xx.xxx

    T14_1

Tarea 15: Configuración de la aplicación de flujo de Java Spark

  1. Vaya a Cubos, haga clic en Crear cubo y cree dos nuevos cubos denominados dataflow-app-avro y dataflow-logs-avro, que se utilizarán para cargar la aplicación java.

  2. Vuelva a comprobar las versiones del entorno java.

    Java

    java 11.0.8 2020-07-14 LTS
Java(TM) SE Runtime Environment 18.9 (build 11.0.8+10-LTS)
Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.8+10-LTS, mixed mode)

    Maven

    Apache Maven 3.5.4 (Red Hat 3.5.4-5)
Maven home: /usr/share/maven
Java version: 11.0.20, vendor: Red Hat, Inc., runtime: /usr/lib/jvm/java-11-openjdk-11.0.20.0.8-3.0.1.el8.x86_64
Default locale: en_US, platform encoding: ANSI_X3.4-1968
OS name: "linux", version: "5.15.0-103.114.4.el8uek.x86_64", arch: "amd64", family: "unix"

  3. Descargue el código de ejemplo y descomprímalo en el entorno local que tiene oci-cli, docker, java y maven: spark-consume-avro-message.zip.

    unzip spark-consume-avro-message.zip
cd spark-consume-avro-message
ls -lrt

    T15_1 T15_1

    Un poco de análisis en el código de proxy para llamar a los tipos de instancias de contenedor A y B.

    Compruebe el archivo de programa principal .src/main/java/example/Example.java.... T15_1

  4. Dado que este programa Java utiliza una biblioteca para manejar la navegación por chispa, necesitamos empaquetar la dependencia para transferirla al flujo de datos. Para ello, utilizaremos Data Flow Dependency Packager, si necesita más detalles, puede ir a Data Flow Dependency Packager.

    El paquete org.apache.spark:spark-avro_2.12:3.2.1 ya está declarado en el archivo packages.txt, solo tiene que empaquetarlo ejecutando:

    docker run --privileged --platform linux/amd64 --rm -v $(pwd):/opt/dataflow  --pull always -it phx.ocir.io/oracle/dataflow/dependency-packager:latest -p 3.8

    T15_1 T15_1 T15_1

  5. Cargue el archivo archive.zip en el cubo de almacenamiento denominado dataflow-app-avro mediante oci-cli.

    oci os object put -bn dataflow-app-avro --file archive.zip --force

  6. Compile, empaquete la aplicación java y cárguela en el cubo de almacenamiento dataflow-app-avro

    ls -lrt
mvn clean install

    T15_1
    ...número reducido de líneas del log de compilación... T15_1

    # upload the JAR file to the storage bucket
oci os object put -bn dataflow-app-avro --file target/consumekafka-1.0-SNAPSHOT.jar --force

    T15_1

  7. Compruebe el cubo de almacenamiento dataflow-app-avro actual y asegúrese de que se parezca a este.

    T15_1

  8. Vaya a Análisis e IA y, a continuación, en Lago de datos, haga clic en Flujo de datos, seleccione el menú izquierdo Puntos finales privados y haga clic en Crear punto final privado.

    • Se necesita el punto final privado, ya que estamos utilizando la subred PRIVATE para las instancias de contenedor y el pool de OCI Streaming.

    • Asegúrese de rellenar las zonas de DNS con el FQDN interno desde las instancias de contenedor de OCI y el pool de OCI Streaming con delimitado por comas.

      T15_1

  9. Vaya a Análisis e IA, en Data Lake, haga clic en Data Flow y, a continuación, haga clic en Crear aplicación.

    T15_1
    T15_1
    T15_1
    T15_1
    T15_1
    T15_1

    • Una vez creado, seleccione el flujo de datos spark-lab-avro y, a continuación, haga clic en Ejecutar para iniciar el programa, normalmente tarda hasta 8 minutos en iniciarse.

      T15_1
      T15_1

  10. Compruebe la aplicación Running Dataflow y abra SparkUI que mostrará los trabajos actuales y la aplicación está funcionando.

    T15_1

    T15_1

    T15_1

Tarea 16: Validar el flujo

Llame a la función y transfiera un mensaje para comprobar que todo el flujo funciona como se esperaba.

  1. Abra el terminal del shell de Linux donde ha creado los mensajes de ejemplo type_a_message.bin y type_b_message.bin y envíe el mensaje. Sustituya la URL de API por el punto final correcto que haya obtenido de la creación del gateway de API.

    cd create_avro_sample/
ls -lrt
curl -X POST -H "Content-Type: application/octet-stream" \
   -d "$(echo -n -e "$(cat type_a_message.bin)")" \
   https://xxxxxxxxxxxxx.apigateway.sa-saopaulo-1.oci.customer-oci.com/

    T16_1

  2. Comprobemos si se ha llamado al tipo de API A comprobando los logs en la instancia de contenedor.

    T16_1 T16_1

Puede repetir el proceso y enviar un archivo type_b_message.bin, y llamará a la instancia de contenedor de tipo B.

T9_1

Agradecimientos

Más recursos de aprendizaje

Explore otros laboratorios en docs.oracle.com/learn o acceda a más contenido de aprendizaje gratuito en el canal YouTube de Oracle Learning. Además, visite education.oracle.com/learning-explorer para convertirse en un explorador de Oracle Learning.

Para obtener documentación sobre los productos, visite Oracle Help Center.