Nota:

Esegui lo streaming dei messaggi AVRO utilizzando Oracle Cloud Infrastructure Streaming e OCI Data Flow con l'elaborazione micro-batch

Introduzione

Nell'attuale panorama basato sui dati, la capacità di elaborare e analizzare flussi di dati in tempo reale è fondamentale per le aziende che mirano a ottenere insight e rispondere rapidamente a condizioni in evoluzione. Le tecnologie di elaborazione dati in streaming sono emerse come soluzioni efficaci per la gestione di flussi di dati continui e a elevato volume. In questa esercitazione esploreremo un approccio innovativo per lo streaming efficiente dei messaggi AVRO utilizzando lo streaming di Oracle Cloud Infrastructure (OCI), associato alla tecnica di elaborazione di micro-batch e ottimizzato con le funzionalità serverless di Oracle Functions sulla base del progetto FN open source.

Introduzione a AVRO e dati in streaming

AVRO, un formato di serializzazione dei dati ampiamente adottato, è noto per la sua efficienza nel rappresentare strutture di dati complesse e la sua compatibilità con vari linguaggi di programmazione. Quando è integrato con le tecnologie di streaming, AVRO consente alle organizzazioni di trasmettere ed elaborare i dati quasi in tempo reale, consentendo loro di estrarre preziose informazioni senza la latenza generalmente associata all'elaborazione in batch.

Streaming OCI: utilizzo di dati in tempo reale

Oracle Cloud Infrastructure (OCI) dispone di una vasta gamma di strumenti per la gestione dei dati nel cloud, mentre lo streaming OCI è un servizio di questo tipo personalizzato per flussi di dati in tempo reale con throughput elevato. Sfruttando OCI Streaming, gli sviluppatori possono creare pipeline di dati scalabili e affidabili che incorporano, elaborano e distribuiscono i flussi di dati in modo efficiente.

Spark gestito di Flusso dati OCI: nessuna soluzione di blocco

Oracle Cloud Infrastructure (OCI) Data Flow è un servizio Apache Spark completamente gestito che esegue attività di elaborazione su set di dati estremamente grandi senza l'infrastruttura da distribuire o gestire.

L'elaborazione microbatch elaborazione microbatch implica l'eliminazione dei flussi di dati in entrata in batch compatti utilizzando il tempo o la dimensione come criteri. Questi batch vengono quindi elaborati come processi più piccoli. A differenza della gestione costante e individuale dei record nell'elaborazione dei flussi, l'elaborazione dei microbatch introduce un po' di ritardo e storage prima dell'elaborazione, che consente un maggiore controllo su cosa fare con i dati. A differenza dell'elaborazione batch tradizionale che affronta i set di Big Data a intervalli regolari, l'elaborazione di microbatch fornisce l'elaborazione quasi in tempo reale e la distribuzione dei risultati.

Sblocco di Synergy: streaming OCI, flusso di dati OCI e Oracle Functions

Questa esercitazione descrive la fusione tra streaming OCI, streaming Spark gestito da Flusso dati OCI e Oracle Functions. Ti guideremo durante tutto il processo di impostazione di una pipeline di dati in streaming end-to-end che include messaggi codificati da AVRO, li elabora in modo efficiente utilizzando le funzionalità di elaborazione micro-batch Spark gestite da OCI Data Flow e introduce l'elaborazione serverless basata sugli eventi con Oracle Functions.

Obiettivo

Utilizza l'elaborazione di microbatch Spark gestita da streaming OCI e Flusso dati OCI per creare una pipeline di elaborazione dati efficiente e in tempo reale utilizzando il formato AVRO.

IMPORTANTE: questa esercitazione è progettata esclusivamente per scopi didattici e di studio. Fornisce agli studenti un ambiente per sperimentare e acquisire esperienza pratica in un'impostazione controllata. È fondamentale notare che le configurazioni e le pratiche di sicurezza utilizzate in questo laboratorio potrebbero non essere adatte agli scenari reali.

Le considerazioni sulla sicurezza per le applicazioni reali sono spesso molto più complesse e dinamiche. Pertanto, prima di implementare qualsiasi tecnica o configurazione qui dimostrata in un ambiente di produzione, è essenziale effettuare una valutazione e un riesame completi della sicurezza. Questa revisione dovrebbe includere tutti gli aspetti della sicurezza, inclusi il controllo dell'accesso, la cifratura, il monitoraggio e la conformità, per garantire che il sistema sia allineato ai criteri e agli standard di sicurezza dell'organizzazione.

La sicurezza deve sempre essere una priorità assoluta quando si passa da un ambiente di laboratorio a una distribuzione reale.

Flusso del processo
T0_1

Architettura di alto livello
T0_1

Prerequisiti - Oracle Cloud Infrastructure

Prerequisiti - Ambiente computer locale

Task 1: Impostazione dei gruppi dinamici

  1. Andare al dominio, fare clic su Gruppi dinamici e creare i gruppi seguenti.

    Nome gruppo: MyFunctions

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

    Nome gruppo: ContainerIntances

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

    Nome gruppo: DataFlowDynamicGroup

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

Task 2: Creare criteri

Task 3: creare bucket di storage e caricare lo schema AVRO

  1. Andare a Bucket, fare clic su Crea bucket e creare un nuovo bucket denominato avro-schema-bucket per memorizzare il file dello schema AVRO.

    T3_1

  2. Ora, seleziona il bucket e fai clic su NOTA del spazio di nomi, ne avremo bisogno in seguito.

    T3_1

  3. Caricare il file user.asvc in questo bucket creato.

    T3_1

Task 4: creare l'argomento di streaming OCI privato

  1. Andare a Analytics e AI, quindi fare clic su Streaming e creare un nuovo flusso denominato FrontDoorTopic.

    T4_0

  2. Selezionare Pool di streaming, fare clic su PrivatePool, quindi fare clic sull'opzione Impostazioni connessione Kafka e su TAKE NOTE dei campi. Sarà necessario farlo in un secondo momento.

    T4_0

Task 5: Creare il TOKEN AUTH

Creare il TOKEN AUTH per l'utente. È necessario utilizzare l'argomento Kafka

  1. Fare clic sull'icona dell'utente in alto a destra, quindi selezionare l'opzione Impostazioni utente.

  2. Fare clic su Autorizza token, quindi generare un nuovo token e su TAKE NOTE del token.

    T4_1

Task 6: Crea registro container

  1. Andare al menu Servizi sviluppatore, fare clic su Registro container e creare i repository privati seguenti.

    Nome repository Tipo
    api-avro-sample_a Privato
    api-avro-sample_b Privato
    fn-recep-avro Privato

  2. Controllare i repository e TAKE NOTE dello spazio di nomi.

    T6_1

  3. Aprire la shell del terminale in cui sono installati CLI e Docker OCI e procedere con il login nel registro. Controlla l'URL corretto per la tua REGION. In questa esercitazione viene utilizzato Brasile East (Sao Paulo) dove l'URL del registro è gru.ocir.io.

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

    T6_1

Task 7: Crea vault OCI

Creare il vault OCI e fornire le variabili necessarie che verranno utilizzate in un secondo momento in questa esercitazione.

  1. Andare a Identifica e sicurezza, fare clic su Errore, quindi su Crea vault.

    T7_1new

  2. Selezionare il nuovo vault e creare le chiavi di cifratura master per tale vault.

    T7_1new

  3. Creare un nuovo segreto denominato AUTH_KEY e incollare la chiave di autenticazione creata in precedenza.

    T7_1new

  4. Ripetere il processo di creazione dei segreti e creare i nuovi segreti seguenti:

    Nome variabile Valore
    KAFKA_BOOTSTRAPSERVER "Il server di avvio dalla configurazione di streaming OCI"
    KAFKA_TOPIC “FrontDoorTopic”
    KAFKA_USERNAME "Il nome utente + l'ID streampool dalla configurazione di streaming OCI"
    AUTH_KEY "Il tuo token AUTH che hai creato sui passi precedenti"

  5. Prendere nota dell'OCID segreto creato per ogni segreto e creare un nuovo file di configurazione.

    • Il file config.properties contiene le variabili di mapping dall'applicazione agli ocid dei segreti del vault. L'applicazione utilizzerà questo file per identificare i segreti del vault da raccogliere durante il runtime.

    • Creare un nuovo file sul computer locale in cui si ha accesso a OCI-CLI:
      Sostituire con l'OCID per ogni Secreat
      Nome file: 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. Andare a Bucket, fare clic su Crea bucket e creare un nuovo bucket denominato config per memorizzare il file config.properties.

  7. Caricare il file config.properties nel bucket di storage config

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

    T7_1new

Task 8: Creare un messaggio AVRO semplice e salvarlo in un file utilizzando il codice python di esempio fornito

  1. Aprire il terminale della shell in cui sono installati CLI OCI, Docker e Python3 e creare un nuovo file AVRO contenente un singolo messaggio in base allo stesso schema AVRO caricato nello storage degli oggetti in precedenza.

    Nota: è necessario controllare la versione Python e utilizzare Python 3.9.16 in questa esercitazione. È possibile che le versioni precedenti non funzionino.

  2. Scarica il codice da qui Create_avro_sample.zip.

  3. Decomprimerlo nella posizione scelta, quindi eseguire il programma per generare un messaggio AVRO di esempio:

    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

Task 9: creare la funzione OCI per ricevere il messaggio AVRO e pubblicare nell'argomento Streaming OCI

  1. Andare a Servizi per sviluppatori, in Funzioni, fare clic su Applicazioni, quindi fare clic su Crea applicazione.

    T9_1

  2. Accedere alla shell del terminale in cui sono installati Docker, CLI OCI, CLI FN ed eseguire i comandi seguenti per inizializzare la funzione.

    Nota: se si sono seguiti i passi, il comando di login Docker è già stato eseguito entro il momento, in caso contrario, procedere con il passo di login Docker nel task Crea registro container.

    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: in questa esercitazione viene utilizzata l'area Brasile East (Sao Paulo), se si utilizza un'area diversa, è necessario modificare le posizioni api-url e registry.

    T9_1

  3. Creare una semplice funzione Hello-world per assicurarsi che tutte le impostazioni siano corrette.

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

    T9_1

  4. Recuperare il codice di esempio della funzione AVRO nel file fn-recep-avro.zip e sostituire il codice globale creato in precedenza. Per funzionare, è necessario ottenere entrambi i file func.py e requirements.txt.

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

    T9_1

  5. Creare il nuovo codice e distribuire la funzione

    fn deploy --app MyReceptionApp

    T9_1

  6. Per richiamare la funzione, è necessario passare un messaggio AVRO come parametro, pertanto verrà utilizzato il file di messaggio AVRO di esempio creato dai passi precedenti. La prima volta che una funzione viene richiamata richiede un po' più tempo da quando deve essere avviata.

    # 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

Task 10: creare un gateway API per esporre la funzione

  1. Nella console fare clic su Servizi per sviluppatori, quindi in Gestione API fare clic su Gateway, quindi fare clic su Crea gateway.

    T10_1

  2. Dopo la creazione, fare clic sull'opzione Distribuzioni, quindi su Crea distribuzione.

    Nome: RecepFunction
    Prefisso percorso: /

    • Nell'autenticazione scegliere Nessuna autenticazione poiché si tratta di un laboratorio semplice e non è stata implementata alcuna autenticazione API. L'obiettivo principale qui è dimostrare una chiamata HTTPS che passa un messaggio AVRO binario tramite API, e ai fini di questo laboratorio, non implementeremo alcun metodo di autenticazione per questo laboratorio semplice.
    • Prima di passare a un ambiente reale, verifica di seguire le procedure di sicurezza consigliate per il gateway API.
    • Per ulteriori dettagli, consulta la sezione relativa alla protezione dei gateway e delle risorse API.

    Instradamento 1: percorso: /

    Metodi: POST
    Tipo di backend: funzioni Oracle
    Applicazione: selezionare la funzione

    T9_1

    T9_1

    T9_1

    T9_1

  3. Controllare se è presente l'endpoint del gateway API e prendere nota.

    T9_1

  4. Aprire il terminale della shell Linux e chiamare il gateway API. Sostituire l'URL dell'API con l'endpoint corretto ottenuto nel passo precedente.

    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

Checkpoint

T9_1

Task 11: Creare l'immagine del contenitore per il tipo di API A

Nota: i codici A e B delle API sono sostanzialmente uguali solo con un messaggio di intestazione diverso per simulare due API diverse.

  1. Recuperare il codice dall'interfaccia API type A e decomprimerlo nel terminale della shell Linux api-avro-sample_a.zip.

  2. Ottiene lo spazio dei nomi del registro dei container disponibile nei passi precedenti e crea la posizione del registro delle applicazioni seguendo il pattern riportato di seguito. L'URL diocir si basa sulla tua area, ad esempio gru.ocir.io per Brasil East(SaoPaulo)

    [ocir url]/[il tuo spazio di nomi]/api-avro-sample_a:latest

  3. Nel terminale della shell Linux, creare ed eseguire il push dell'immagine docker per questa 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

Task 12: Creare l'immagine del contenitore per il tipo di API B

  1. Recuperare il codice dall'interfaccia API type B e decomprimerlo nel terminale della shell Linux api-avro-sample_b.zip.

  2. Ottiene lo spazio dei nomi del registro dei container disponibile nei passi precedenti e crea la posizione del registro delle applicazioni seguendo il pattern riportato di seguito. L'URL diocir si basa sulla tua area, ad esempio gru.ocir.io per Brasil East(SaoPaulo)

    [ocir url]/[il tuo spazio di nomi]/api-avro-sample_b:latest

  3. Nel terminale della shell Linux, creare ed eseguire il push dell'immagine docker per questa 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. Verificare nella pagina del registro container se il push dell'immagine è riuscito.

    T10_1

Task 13: distribuire le API nel servizio container

  1. Andare a Servizi per sviluppatori, Istanze contenitore e fare clic su Crea istanza contenitore.

    T13_1

    T13_1

  2. Ripetere il passo 1 per api-type-b e selezionare l'immagine corretta per l'API TYPE B.

    1. Andare a Servizi per sviluppatori, Istanze contenitore, quindi fare clic su Crea istanza contenitore e ripetere i passi per distribuire il tipo di API B.

    2. Recuperare l'indirizzo FQDN interno dalle istanze del contenitore.

      T14_1

      • Fare clic sull'istanza del contenitore e prendere nota di ogni indirizzo FQDN interno.

      T14_1

    3. Andare a Identifica e sicurezza, fare clic su Errore, selezionare il VAULT e creare due nuovi segreti.

      Nome segreto Valore
      TIPO_API_A_URL Incolla l'indirizzo privato interno FQDN per l'API di tipo A
      API_TYPE_B_URL Incolla l'indirizzo privato interno FQDN per l'API di tipo B

      Prendi nota di ogni OCID segreto

      Il vault dovrebbe avere questo aspetto ora:

      T14_1

    4. Modificare il file config.properties caricato nel bucket di storage config e aggiungere nuove voci per l'OCID segreto

      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

      Il file avrà l'aspetto seguente:
      T14_1

      T14_1

Task 14: eseguire il test delle interfacce API utilizzando create_avro_sample.py

  1. Andare al terminale della shell Linux in cui è stato salvato il file create_avro_sample.py da Task 7 e creare alcuni nuovi messaggi per eseguire il test delle chiamate API. Creazione di due nuovi file AVRO con ID diversi (1010 e 1020) che verranno utilizzati come filtro all'interno del programma 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. Chiamare che le API che superano il messaggio AVRO da sottoporre a test funzionino correttamente. Andare alla pagina Istanze contenitore e ottenere l'indirizzo FQDN interno per ciascuna API api-type-a e api-type-b. Ricordarsi di sostituire l'URL riportato di seguito per l'indirizzo FQDN interno corrispondente dalle 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

Task 15: impostazione dell'applicazione di streaming Java Spark

  1. Andare a Bucket, fare clic su Crea bucket e creare due nuovi bucket denominati dataflow-app-avro e dataflow-logs-avro, che verranno utilizzati per caricare l'applicazione java.

  2. Controllare le versioni dell'ambiente 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. Scaricare il codice di esempio ed estrarlo dall'ambiente locale con oci-cli, docker, java e maven: spark-consume-avro-message.zip.

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

    T15_1 T15_1

    Approfondimento del codice proxy per richiamare i tipi di istanze contenitore A e B.

    Verificare il file del programma principale .src/main/java/example/Example.java.... T15_1

  4. Poiché questo programma Java utilizza una libreria per gestire spark-avro, è necessario raggruppare l'area depedenza per passare al flusso di dati. Per questo motivo, verrà utilizzato Data Flow Dependency Packager, se sono necessari ulteriori dettagli, è possibile andare a Data Flow Dependency Packager.

    Il pacchetto org.apache.spark:spark-avro_2.12:3.2.1 è già stato dichiarato nel file packages.txt, è sufficiente eseguirlo eseguendo:

    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. Caricare il file archive.zip nel bucket di storage denominato dataflow-app-avro, utilizzando oci-cli.

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

  6. Compila, crea il package dell'applicazione java e caricala nel bucket di storage dataflow-app-avro

    ls -lrt
mvn clean install

    T15_1
    ...numero ridotto di righe del log di compilazione... 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. Controllare il bucket di storage dataflow-app-avro corrente e assicurarsi che sia simile al seguente.

    T15_1

  8. Andare a Analitica e AI, quindi in Data lake fare clic su Flusso dati, selezionare il menu di sinistra endpoint privati e fare clic su Crea endpoint privato.

    • L'endpoint privato è necessario poiché stiamo utilizzando la subnet PRIVATE per le istanze contenitore e il pool di streaming OCI.

    • Assicurarsi di riempire le zone DNS con il nome FQDN interno delle istanze contenitore OCI e del pool di streaming OCI con valori delimitati da virgole.

      T15_1

  9. Andare a Analitica e AI, quindi in Data lake fare clic su Flusso di dati, quindi fare clic su Crea applicazione.

    T15_1
    T15_1
    T15_1
    T15_1
    T15_1
    T15_1

    • Una volta creato, selezionare il flusso di dati spark-lab-avro, quindi fare clic su Esegui per avviare il programma, in genere sono necessari fino a 8 minuti per l'avvio.

      T15_1
      T15_1

  10. Controllare l'applicazione del flusso di dati in esecuzione e aprire il file SparkUI che mostra i job correnti e l'applicazione funziona.

    T15_1

    T15_1

    T15_1

Task 16: convalidare il flusso

Chiamare la funzione e passare un messaggio per verificare che tutto il flusso funzioni come previsto.

  1. Aprire il terminale della shell Linux in cui sono stati creati i messaggi di esempio type_a_message.bin e type_b_message.bin e inviare il messaggio. Sostituisci l'URL dell'API con l'endpoint corretto ottenuto dalla creazione del gateway 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. Verificare se il tipo di API A è stato richiamato controllando i log sull'istanza del contenitore.

    T16_1 T16_1

È possibile ripetere il processo e inviare un file type_b_message.bin, che chiamerà l'istanza del contenitore di tipo B.

T9_1

Approvazioni

Altre risorse di apprendimento

Esplora altri laboratori su docs.oracle.com/learn o accedi a contenuti di formazione gratuiti sul canale YouTube di Oracle Learning. Inoltre, visitare education.oracle.com/learning-explorer per diventare Explorer di Oracle Learning.

Per la documentazione sul prodotto, visitare il sito Oracle Help Center.