Nota:

Questa esercitazione richiede l'accesso a Oracle Cloud. Per iscriversi a un account gratuito, vedere Inizia a utilizzare Oracle Cloud Infrastructure Free Tier.
Utilizza valori di esempio per le credenziali, la tenancy e i compartimenti Oracle Cloud Infrastructure. Al termine del laboratorio, sostituire questi valori con quelli specifici del tuo ambiente cloud.

Ottimizza l'analisi dei dati utilizzando Oracle Cloud Infrastructure Streaming e Oracle Database

Introduzione

Oracle Cloud Infrastructure Streaming è un servizio di streaming altamente scalabile e ad alta disponibilità su Oracle Cloud Infrastructure (OCI). Il servizio Stream è completamente serverless e compatibile con Apache Kafka.

Oracle SQL access to Kafka è un pacchetto Oracle PL/SQL e un preprocessore di tabella esterno. Consente a Oracle Database di leggere ed elaborare eventi e messaggi dagli argomenti Kafka sotto forma di viste o tabelle in Oracle Database. Quando i dati di Kafka si trovano in una tabella Oracle Database o sono visibili da una vista di Oracle Database, è possibile eseguirne una query completa con la potenza completa di Oracle PL/SQL, proprio come qualsiasi altro dato in Oracle Database.

Quando si recuperano i dati con l'accesso Oracle SQL alle viste abilitate per Kafka, i dati non vengono resi persistenti in Oracle Database. Tuttavia, quando si utilizza l'accesso Oracle SQL alle tabelle basate su Kafka, l'accesso viene reso persistente in Oracle Database. Pertanto l'accesso Oracle SQL a Kafka offre agli sviluppatori flessibilità completa sulla persistenza dei dati in streaming su Oracle Database. Per una panoramica e un caso d'uso corretti per l'accesso a Kafka a Oracle SQL, consultare questo post del blog, Integrating Data-in-Motion with Data-at-Rest using Oracle SQL Access to Kafka Viewer.

In breve, l'accesso Oracle SQL a Kafka consente di elaborare i dati in movimento con i dati in archivio (all'interno delle tabelle di Oracle Database) mediante Oracle PL/SQL. Pertanto, le applicazioni, ad esempio, Java Database Connectivity (JDBC) possono elaborare eventi in tempo reale da Kafka e dei dati Oracle Database critici, all'interno di transazioni di Oracle Database che offrono garanzie ACID. Questa operazione non è facilmente possibile quando un'applicazione recupera separatamente eventi e dati Kafka da Oracle Database.

Vantaggi

I clienti possono utilizzare l'accesso Oracle SQL a Kafka per eseguire job di analitica dei flussi in tempo reale mediante Oracle PL/SQL leggendo direttamente i dati dal servizio Stream, senza mai spostarli in un data store esterno.
I clienti possono anche utilizzare l'accesso Oracle SQL a Kafka solo per lo spostamento dei dati dal servizio Stream a Oracle Database senza alcun tipo di elaborazione.
L'operazione di elaborazione del flusso può essere eseguita nel contesto di una transazione Oracle ACID controllata dall'applicazione.
L'accesso Oracle SQL a Kafka agisce solo come consumatore Kafka e non come produttore Kafka. L'intera gestione degli offset è gestita da OSaK. Memorizza queste informazioni nelle tabelle dei metadati in Oracle Database. Pertanto, l'accesso Oracle SQL a Kafka abilita la semantica dell'elaborazione di una volta esatta in quanto può eseguire il commit di partition offset per Kafka e il servizio Stream e dei dati dell'applicazione in una singola transazione Oracle Database conforme all'ACID in Oracle Database. Ciò elimina la perdita o la rilettura dei record di streaming.

Casi d'uso

Immagina qualsiasi caso d'uso in cui desideri correlare o unire i tuoi dati in streaming con i tuoi dati relazionali, ad esempio:

Vuoi combinare i dati di streaming provenienti dai tuoi chatty dispositivi IoT (molto probabilmente presenti on premise ai tuoi clienti) con le informazioni sui clienti rilevanti, memorizzati nella tua versione con Oracle Database relazionale per la verità sorgente.
Si desidera calcolare le medie di spostamento esponenziali precise per i prezzi delle azioni che si spostano nel servizio Stream. È necessario farlo con semantica esattamente una volta. Vuoi combinare questi dati con informazioni statiche su tale stock, ad esempio il nome, l'ID azienda, il limite del mercato e così via, memorizzato nel tuo Oracle Database.

Tenere presente che l'accesso Oracle SQL a Kafka è un package PL/SQL che deve essere installato manualmente nell'host Oracle Server, può funzionare solo con le installazioni Oracle Database autogestite (in locale o nel cloud). Non può funzionare con offerte Oracle Database serverless come Oracle Autonomous Database su Oracle Cloud Infrastructure (OCI).

Poiché il servizio Stream è compatibile con Kafka, funziona in modo trasparente con l'accesso Oracle SQL a Kafka. Dall'accesso Oracle SQL al punto di vista Kafka, un pool di flussi di servizi Stream è un cluster Kafka e un flusso di servizi Stream è un argomento nel cluster Kafka.

In questa esercitazione verrà illustrato con quanta facilità è possibile integrare Oracle SQL Access in Kafka con il servizio Stream.

Nota: se si ha già familiarità con l'accesso Oracle SQL a Kafka, il servizio Stream e Kafka e si desidera utilizzare l'accesso Oracle SQL a Kafka con il servizio Stream, è possibile passare direttamente a Imposta cluster Oracle Cloud Infrastructure Streaming per l'accesso Oracle SQL al passo Kafka 2.1. Abilità nel resto dell'esercitazione in base alle esigenze.

Prerequisiti

Account o tenancy OCI con autorizzazione per creare e utilizzare il servizio Stream.
Apache Kafka cluster 0.10.2 o versioni successive. Il servizio Stream soddisfa questo requisito.
Oracle Database 12.2 o versione successiva in cui verrà installato l'accesso Oracle SQL a Kafka (la versione 19c viene utilizzata in questa esercitazione). Stiamo utilizzando Oracle Linux come sistema operativo di piattaforma per Oracle Database. Le istruzioni per e grandi rimangono invariate per altre piattaforme.
Java 8 o versione successiva installato sull'host Oracle Database.

Oracle Cloud Infrastructure Streaming e Oracle SQL Access to Kafka Integration

Crea un pool di flussi e accedi a Oracle Cloud Infrastructure

Eseguire il login all'account/tenancy OCI e impostare un pool di flussi del servizio di streaming denominato StreampoolForOsak e il flusso denominato StreamForOsak come indicato di seguito.
Ora creiamo un flusso denominato StreamForOsak nello streampool StreampoolForOsak che abbiamo appena creato.

Per queste creazioni di risorse, puoi utilizzare qualsiasi compartimento esistente. Abbiamo pertanto creato un nuovo compartimento denominato OssOsakIntegration e tutte le risorse si trovano nello stesso compartimento.

Nella terminologia del servizio Stream, gli argomenti Kafka sono denominati Streams. Pertanto, dall'accesso Oracle SQL al punto di vista Kafka il flusso StreamForOsak è un argomento Kafka con tre partizioni.

Ora vengono creati i flussi del servizio di streaming. In seguito, i termini del servizio di streaming e Kafka sono intercambiabili. Analogamente, i termini Stream e Kafka Topic sono intercambiabili.

Crea un utente, un gruppo e i criteri in Oracle Cloud Infrastructure

Se si dispone già di un utente esistente con le autorizzazioni appropriate per l'utilizzo del servizio Stream, il passo 2 può essere saltato completamente.

Per utilizzare i flussi di servizi di streaming con l'accesso Oracle SQL a Kafka, è necessario creare un nuovo utente OCI specifico. Per questo scopo, creiamo un nuovo utente con il nome utente OssOsakUser dalla console Web OCI, come indicato di seguito.
Per consentire all'utente OssOsakUser di autenticarsi con il servizio Stream (mediante le API Kafka), è necessario creare un token di autenticazione per questo nuovo utente, come descritto di seguito.

Dopo aver generato il token, si ottiene prima e l'unica possibilità di visualizzare e copiare il token di autenticazione. Quindi, copiare l'autenticazione e conservarla in modo sicuro in un luogo da cui è possibile accedervi in seguito. È necessaria nei passi successivi, in particolare quando configuri il cluster Kafka per l'accesso a Kafka tramite Oracle SQL. L'accesso SQL Oracle a Kafka utilizzerà questo nome utente, ovvero OssOsakUser e il relativo token di autenticazione per accedere al servizio Stream, utilizzando le API consumer Kafka internamente.
Inoltre, OssOsakUser deve disporre del set di privilegi appropriato in IAM OCI per accedere al cluster del servizio Stream.

In OCI, gli utenti acquisiscono i privilegi con l'ausilio di criteri assegnati ai gruppi di utenti di cui fanno parte. Pertanto, è necessario creare un gruppo per OssOsakUser seguito da un criterio per tale gruppo.

Creare un gruppo utenti come mostrato di seguito:
Aggiungere l'utente OssOsakUser al gruppo utente OssOsakUserGroup.
Per autorizzare OssOsakUser a utilizzare il flusso del servizio Stream, in particolare per pubblicare e leggere i messaggi, è necessario creare il seguente criterio nella tenancy. Questo criterio, una volta creato, concede privilegi a tutti gli utenti del gruppo OssOsakUserGroup. Poiché l'utente OssOsakUser si trova nello stesso gruppo, acquisisce gli stessi privilegi.

Lo snippet di testo per lo stesso criterio precedente è il seguente.
```
Allow group OssOsakUserGroup to manage streams in compartment OssOsakIntegration  
Allow group OssOsakUserGroup to manage stream-push in compartment OssOsakIntegration   
Allow group OssOsakUserGroup to manage stream-pull in compartment OssOsakIntegration  
Allow group OssOsakUserGroup to manage stream-family in compartment OssOsakIntegration
```

Installare Oracle SQL Access to Kafka su un host Oracle Database

L'accesso Oracle SQL al kit Kafka è disponibile nell'ambito di un SQL Developer ampiamente utilizzato.

Utilizzare il collegamento SQL Developer ufficiale per scaricare la versione più recente di SQL Developer. A partire dalla scrittura di questa esercitazione, la versione più recente di SQL Developer è la 20.4.

Assicurarsi di scaricare Oracle SQL Developer per la stessa piattaforma dell'host Oracle Database. Sulla nostra casella di sviluppo scarichiamo Oracle SQL Developer per Oracle Linux RPM (come il nostro Oracle Database è in esecuzione sulla piattaforma Oracle Linux).

Una volta completato il download, estrarre il contenuto del file RPM/zip di Oracle SQL Developer in qualsiasi directory/cartella utilizzando il comando unzip o tar.
```
tar xvf sqldeveloper-20.4.0.379.2205-20.4.0-379.2205.noarch.rpm
```
Andare alla directory in cui sono stati estratti i contenuti di Oracle SQL Developer e cercare l'accesso a Oracle SQL al file ZIP Kafka denominato orakafka.zip come descritto di seguito.
```
$ find . -name 'orakafta*'  
./sqldeveloper/orakafta  
./sqldeveloper/orakafta/orakafta.zip
```
L'accesso Oracle SQL al kit Kafka si trova nel file orakafka.zip. Non ci interessa il resto dei contenuti di SQL developer per questa demo. Per quanto riguarda l'installazione e l'uso dell'accesso Oracle SQL a Kafka, è necessario disporre di orakafka.zip.
Copiare il file orakafka.zip nell'host Oracle Database utilizzando il comando scp o i client FTP basati sulla GUI come FileZilla.

SSH nel nodo Oracle Database e sposta orakafka.zip in /home/oracle(directory home dell'utente oracle) con mv_ command.

Per il resto delle istruzioni di installazione dell'accesso Oracle SQL a Kafka, è necessario passare all'utente oracle sull'host Oracle Database.

Assicurarsi che la directory di lavoro corrente sia /home/oracle. Il file orakafka.zip si trova già nella stessa directory.

[opc@dbass ~]$ sudo su - oracle  
Last login: Sat Feb 20 09:31:12 UTC 2021  
[oracle@dbass ~]$ pwd  
/home/oracle  
[oracle@dbass ~]$ ls -al  
total 3968  
drwx------ 6 oracle oinstall    4096 Feb 19 17:39 .  
drwxr-xr-x 5 root   root        4096 Feb 18 15:15 ..  
-rw------- 1 oracle oinstall    4397 Feb 20 09:31 .bash_history  
-rw-r--r-- 1 oracle oinstall      18 Nov 22  2019 .bash_logout  
-rw-r--r-- 1 oracle oinstall     203 Feb 18 15:15 .bash_profile  
-rw-r--r-- 1 oracle oinstall     545 Feb 18 15:20 .bashrc  
-rw-r--r-- 1 oracle oinstall     172 Apr  1  2020 .kshrc  
drwxr----- 3 oracle oinstall    4096 Feb 19 17:37 .pki  
drwxr-xr-x 2 oracle oinstall    4096 Feb 18 15:20 .ssh  
-rw-r--r-- 1 oracle oinstall 4002875 Feb 19 17:38 orakafka.zip

Estrarre o decomprimere orakafka.zip. Verrà creata una nuova directory denominata orakafka-<version> con il contenuto estratto. Nel nostro caso, è orakafka-1.2.0 come segue:
```
[oracle@dbass ~]$ unzip orakafka.zip  
Archive:  orakafka.zip 
creating: orakafka-1.2.0/  
```

estrazione: orakafka-1.2.0/kit_version.txt
generazione di informazioni: orakafka-1.2.0/orakafka_distro_install.sh
estrazione: orakafka-1.2.0/orakafka.zip
generazione di informazioni: orakafka-1.2.0/README

6. Now we follow the instructions found in the _**orakafka-1.2.0/README**_ for the setup of Oracle SQL access to Kafka. We follow _simple install_ for single-instance Oracle Database. 

This README doc has instructions for Oracle SQL access to Kafka installation on Oracle Real Application Clusters (Oracle RAC) as well. By and large, in the case of Oracle RAC, we need to replicate the following steps on all nodes of Oracle RAC. Please follow the README for details.

[oracle@dbass ~]$ cd orakafka-1.2.0/
[oracle@dbass orakafka-1.2.0]$ ls -al
total 3944
drwxrwxr-x 2 oracle oinstall 4096 Feb 20 09:12 .
drwx—— 6 oracle oinstall 4096 Feb 19 17:39 ..
-rw-r–r– 1 oracle oinstall 6771 Oct 16 03:11 README
-rw-r–r– 1 oracle oinstall 5 Oct 16 03:11 kit_version.txt
-rw-rw-r– 1 oracle oinstall 3996158 Oct 16 03:11 orakafka.zip
-rwxr-xr-x 1 oracle oinstall 17599 Oct 16 03:11 orakafka_distro_install.sh

tar xvf sqldeveloper-20.4.0.379.2205-20.4.0-379.2205.noarch.rpm

7. As per _./orakafka-1.2.0/README_, we install Oracle SQL access to Kafka on the Oracle Database host with the help of _./orakafka-1.2.0/orakafka\_distro\_install.sh_ script. Argument -p lets us specify the location or base directory for the Oracle SQL access to Kafka installation on this host. 

   We choose the newly created empty directory named ora\_kafka\_home as the OSaK base directory on this host. So the full path of the OSaK base directory will be _/home/oracle/ora\_kafka\_home_.

[oracle@dbass ~]$ ./orakafka-1.2.0/orakafka_distro_install.sh -p ./ora_kafka_home/

Step Create Product Home::
————————————————————–
…../home/oracle/ora_kafka_home already exists..
Step Create Product Home: completed.
PRODUCT_HOME=/home/oracle/ora_kafka_home

Step Create app_data home::
————————————————————–
….. creating /home/oracle/ora_kafka_home/app_data and subdirectories
……Generated CONF_KIT_HOME_SCRIPT=/home/oracle/ora_kafka_home/app_data/scripts/configure_kit_home.sh
……Generated CONF_APP_DATA_HOME_SCRIPT=/home/oracle/ora_kafka_home/configure_app_data_home.sh
Step Create app_data home: completed.
APP_DATA_HOME=/home/oracle/ora_kafka_home/app_data

Step unzip_kit::
————————————————————–
…..checking for existing binaries in /home/oracle/ora_kafka_home/orakafka
…..unzip kit into /home/oracle/ora_kafka_home/orakafka
Archive: /home/oracle/orakafka-1.2.0/orakafka.zip
creating: /home/oracle/ora_kafka_home/orakafka/
extracting: /home/oracle/ora_kafka_home/orakafka/kit_version.txt
inflating: /home/oracle/ora_kafka_home/orakafka/README
creating: /home/oracle/ora_kafka_home/orakafka/doc/
inflating: /home/oracle/ora_kafka_home/orakafka/doc/README_INSTALL
creating: /home/oracle/ora_kafka_home/orakafka/jlib/
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/osakafka.jar
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/kafka-clients-2.5.0.jar
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/slf4j-simple-1.7.28.jar
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/lz4-java-no-jni-1.7.1.jar
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/snappy-java-no-jni-1.1.7.3.jar
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/zstd-no-jni-1.4.4-7.jar
inflating: /home/oracle/ora_kafka_home/orakafka/jlib/slf4j-api-1.7.30.jar
creating: /home/oracle/ora_kafka_home/orakafka/bin/
inflating: /home/oracle/ora_kafka_home/orakafka/bin/orakafka_stream.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/orakafka.sh
creating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/removeuser_cluster.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/setup_all.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/remove_cluster.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/verify_install.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/add_cluster.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/config_util.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/test_views.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/install.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/set_java_home.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/list_clusters.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/adduser_cluster.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/uninstall.sh
inflating: /home/oracle/ora_kafka_home/orakafka/bin/scripts/test_cluster.sh
creating: /home/oracle/ora_kafka_home/orakafka/conf/
inflating: /home/oracle/ora_kafka_home/orakafka/conf/orakafka.properties.template
creating: /home/oracle/ora_kafka_home/orakafka/sql/
inflating: /home/oracle/ora_kafka_home/orakafka/sql/orakafkatab.plb
inflating: /home/oracle/ora_kafka_home/orakafka/sql/catnoorakafka.sql
inflating: /home/oracle/ora_kafka_home/orakafka/sql/catorakafka.sql
inflating: /home/oracle/ora_kafka_home/orakafka/sql/pvtorakafkaus.plb
inflating: /home/oracle/ora_kafka_home/orakafka/sql/orakafka_pkg_uninstall.sql
inflating: /home/oracle/ora_kafka_home/orakafka/sql/orakafkab.plb
inflating: /home/oracle/ora_kafka_home/orakafka/sql/pvtorakafkaub.plb
inflating: /home/oracle/ora_kafka_home/orakafka/sql/orakafka_pkg_install.sql
inflating: /home/oracle/ora_kafka_home/orakafka/sql/orakafkas.sql
creating: /home/oracle/ora_kafka_home/orakafka/lib/
inflating: /home/oracle/ora_kafka_home/orakafka/lib/libsnappyjava.so
inflating: /home/oracle/ora_kafka_home/orakafka/lib/libzstd-jni.so
inflating: /home/oracle/ora_kafka_home/orakafka/lib/liblz4-java.so
Step unzip_kit: completed.

Successfully installed orakafka kit in /home/oracle/ora_kafka_home

8. Configure JAVA\_HOME for Oracle SQL access to Kafka. We find the Java path on the node as follows:

[oracle@dbass ~]$ java -XshowSettings:properties -version 2>&1 > /dev/null | grep 'java.home'
java.home = /usr/java/jre1.8.0_271-amd64
[oracle@dbass ~]$ export JAVA_HOME=/usr/java/jre1.8.0_271-amd64

   To set Java home for OSaK, we use the script _/home/oracle/ora\_kafka\_home/orakafka/bin/orakafka.sh_ script, with _set\_java\_home_ option.

[oracle@dbass bin]$ pwd
/home/oracle/ora_kafka_home/orakafka/bin
[oracle@dbass bin]$ ./orakafka.sh set_java_home -p $JAVA_HOME

Step1: Check for valid JAVA_HOME
————————————————————–
Found /usr/java/jre1.8.0_271-amd64/bin/java, JAVA_HOME path is valid.
Step1 succeeded.

Step2: JAVA version check
————————————————————–
java version “1.8.0_271”
Java(TM) SE Runtime Environment (build 1.8.0_271-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.271-b09, mixed mode)
Java version >= 1.8
Step2 succeeded.

Step3: Creating configure_java.sh script
————————————————————–
Wrote to /home/oracle/ora_kafka_home/app_data/scripts/configure_java.sh
Step3 succeeded.

Successfully configured JAVA_HOME in /home/oracle/ora_kafka_home/app_data/scripts/configure_java.sh

The above information is written to /home/oracle/ora_kafka_home/app_data/logs/set_java_home.log.2021.02.20-04.38.23

[oracle@dbass bin]$

9. Verify installation of OSaK with _verify\_install_ option of script _orakafka.sh_ as follows:

[oracle@dbass ~]$ cd ora_kafka_home/
[oracle@dbass ora_kafka_home]$ cd orakafka/bin/
[oracle@dbass bin]$ ./orakafka.sh verify_install
Controllare la proprietà di tutti i file/dirs - passato
Controllare i privilegi della directory - passato
Controllare i file eseguibili previsti - passato

Le informazioni sopra riportate sono scritte su /home/oracle/ora_kafka_home/app_data/logs/verify_install.log.2021.02.19-18.10.15

### Set Up Oracle Cloud Infrastructure Stream Clusters for Oracle SQL Access to Kafka

1. Under _ora\_kafka\_home_, that is _/home/oracle/ora\_kafka\_home_, we have two more READMEs as follows:

[oracle@dbass ~]$ trova ora_kafka_home/ -name "README*"
ora_kafka_home/orakafka/README
ora_kafka_home/orakafka/doc/README_INSTALL

   _~/ora\_kafka\_home/orakafka/README_ is the readme for this release of OSaK that we have installed. Please read through this readme.

   And _ora\_kafka\_home/orakafka/doc/README\_INSTALL_ is the README for the actual setup of a Kafka cluster for Oracle SQL access to Kafka. The rest of the steps 2 onwards below follow this README by and large. As per the same, we will leverage _~/ora\_kafka\_home/orakafka/bin/orakafka.s_h script, for adding a Kafka cluster, adding an Oracle Database user for using the Kafka cluster in the next steps.

2. Add our Stream service stream pool to Oracle SQL access to Kafka.

   As mentioned earlier, from the Oracle SQL access to Kafka point of view, the Stream service stream pool is a Kafka cluster.

   We use the _add\_cluster_ option of the _orakafka.sh_ script to add the cluster to OSaK. We use the _\-c_ argument to name the cluster as _kc1_.

[oracle@dbass bin]$ pwd
/home/oracle/ora_kafka_home/orakafka/bin
[oracle@dbass bin]$ ./orakafka.sh add_cluster -c kc1

Step1: creazione della directory di configurazione del cluster del file system
---------------------
Creazione della directory del cluster del file system completata.
Configurare le proprietà di sicurezza all'indirizzo /home/oracle/ora_kafka_home/app_data/clusters/KC1/conf/orakafka.properties.
Step1 riuscito.

Step2: Genera DDL per la creazione della directory DB di configurazione cluster
---------------------
Eseguire lo script SQL seguente durante la connessione come sysdba
per creare la directory di database di configurazione cluster:
@/home/oracle/ora_kafka_home/app_data/scratch/orakafka_create_KC1_CONF_DIR.sql
Step2 ha generato correttamente lo script.

****SUMMARY*****

Task TODO:

Configurare le proprietà di sicurezza all'indirizzo /home/oracle/ora_kafka_home/app_data/clusters/KC1/conf/orakafka.properties
Eseguire il seguente SQL durante la connessione come sysdba:
@/home/oracle/ora_kafka_home/app_data/scratch/orakafka_create_KC1_CONF_DIR.sql

Le informazioni di cui sopra sono scritte in /home/oracle/ora_kafka_home/app_data/logs/add_cluster.log.2021.02.20-05.23.30

[oracle@dbass bin]$

   We get two TODO tasks as per output from the above execution of the _add\_cluster_ command. 

   1. For the first task, we add security properties for our Stream service stream to `/home/oracle/ora_kafka_home/app_data/clusters/KC1/conf/orakafka.properties`. 

      Where to get Kafka-compatible security configs for the Stream service cluster also known as (AKA) streampool? We get these security configs from the OCI web console as shown below.

      Please take note of the bootstrap server endpoint(_cell-1.streaming.ap-mumbai-1.oci.oraclecloud.com:9092_ from the screenshot below) as well. It is not related to the security configs of the cluster, but we need it in later steps for connecting to our Stream service and Kafka cluster. 

      ![](images/bsend.png " ")

      We write the above config values to OSaK in the following format, to `/home/oracle/ora_kafka_home/app_data/clusters/KC1/conf/orakafka.properties` file using any text editor like say vi. 

      ![](images/osakosssec.png " ")

      For clarity, we have the same configs in text format as follows:

      ```
      security.protocol=SASL_SSL  
      sasl.mechanism=PLAIN  
      sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="<YOUR_TENANCY_NAME>/<YOUR_OCI_USERID_HERE>/<YOUR_OCI_STREAMPOOL_OCID>" password=" YOUR_AUTH_TOKEN";  
      sasl.plain.username="<YOUR_TENANCY_NAME>/<YOUR_OCI_USERID_HERE>/<YOUR_OCI_STREAMPOOL_OCID>"  
      sasl.plain.password ="YOUR_AUTH_TOKEN";
      ```

      **Note:** default Kafka Consumer Configs are already pre-populated in _orakafka.properties_ for this cluster. If there is a need, one can modify these configs.

   2. For the second TODO task, do SQL script execution as follows: 

      ```
      [oracle@dbass ~]$ sqlplus / as sysdba  
      SQL*Plus: Release 19.0.0.0.0 - Production on Sat Feb 20 05:37:28 2021  
      Version 19.9.0.0.0   
      Copyright (c) 1982, 2020, Oracle.  All rights reserved.
      Connected to:  
      Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
      Version 19.9.0.0.0   
      SQL> alter session set container=p0;  
      Session altered.  
      SQL> @/home/oracle/ora_kafka_home/app_data/scratch/orakafka_create_KC1_CONF_DIR.sql  
      Creating database directory "KC1_CONF_DIR"..  
      Directory created.  
      The above information is written to /home/oracle/ora_kafka_home/app_data/logs/orakafka_create_KC1_CONF_DIR.log  
      Disconnected from Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
      Version 19.9.0.0.0
      ```

      Take note of directory object _KC1\_CONF\_DIR_ created by the above SQL script.

      **Note:** As dictated in add\_cluster output, we run the SQL script as a _SYSDBA_ user, inside the PDB of our interest (_p0_ PDB here).

   3. To make sure that the cluster configuration is working, we can leverage the _test\_cluster_ option of script _orakafka.sh_.

      We pass cluster name with -c argument and bootstrap server with -b argument. We have from bootstrap server info from step 4.2.1.

      ```
      [oracle@dbass bin]$ pwd  
      /home/oracle/ora_kafka_home/orakafka/bin  
      [oracle@dbass bin]$ ./orakafka.sh test_cluster -c kc1 -b cell-1.streaming.ap-mumbai-1.oci.oraclecloud.com:9092  
      Kafka cluster configuration test succeeded.  
         
      Kafka cluster configuration test - "KC1"  
      ------------------------------------------------------  
      KAFKA_BOOTSTRAP_SERVERS=cell-1.streaming.ap-mumbai-1.oci.oraclecloud.com:9092  
      LOCATION_FILE="/home/oracle/ora_kafka_home/app_data/clusters/KC1/logs/orakafka_stream_KC1.loc"  
      ora_kafka_operation=metadata  
      kafka_bootstrap_servers=cell-1.streaming.ap-mumbai-1.oci.oraclecloud.com:9092  
         
      List topics output:  
      StreamForOsak,3  
      ...  
      For full log, please look at /home/oracle/ora_kafka_home/app_data/clusters/KC1/logs/test_KC1.log.2021.02.20-09.01.12  
      Test of Kafka cluster configuration for "KC1" completed.  
         
      The above information is written to /home/oracle/ora_kafka_home/app_data/logs/test_cluster.log.2021.02.20-09.01.12  
         
      [oracle@dbass bin]$
      ```

      As you can see our Kafka cluster configuration test succeeded! We get the list of topics, AKA the Stream service streams in the output here. Topic names are followed by the number of partitions for that stream. Here we have the number of partitions as three, as the Stream service stream StreamForOsak has exactly three partitions.

3. Configure an Oracle Database user for the added Kafka cluster.

   We already have created an Oracle Pluggable Databases (PDB) level Oracle Database user with username books\_admin, for the PDB named _p0_.  We use the _adduser\_cluster_ option to grant required permissions for user books\_admin on Kafka cluster _kc1_. 

[oracle@dbass bin]$ ./orakafka.sh adduser_cluster -c kc1 -u books_admin

   
Step1: Generate DDL to grant permissions on cluster configuration directory to "BOOKS_ADMIN"
--------------------------------------------------------------------------------------------

Execute the following script while connected as sysdba  
to grant permissions on cluster conf directory :  
 @/home/oracle/ora_kafka_home/app_data/scratch/orakafka_adduser_cluster_KC1_user1.sql  
Step1 successfully generated script.  
   
***********SUMMARY************  
   
TODO tasks:  
 
1. Execute the following SQL while connected as sysdba:  
      @/home/oracle/ora_kafka_home/app_data/scratch/orakafka_adduser_cluster_KC1_user1.sql  
      
   The above information is written to /home/oracle/ora_kafka_home/app_data/logs/adduser_cluster.log.2021.02.20-10.45.57  
      
   [oracle@dbass bin]$

Come nel passo precedente, è presente un task TODO da eseguire.

Secondo l'output, esiste uno script SQL generato automaticamente. È necessario eseguire questo script (come sysdba) per concedere all'utente Oracle Database le autorizzazioni books_admin sulla directory di configurazione del cluster per il cluster kc1.

Nel nostro caso, questa directory è /home/oracle/ora_kafka_home/app_data/clusters/KC1.

[oracle@dbass ~]$ sqlplus / as sysdba  
SQL*Plus: Release 19.0.0.0.0 - Production on Sat Feb 20 10:56:17 2021  
Version 19.9.0.0.0  
Copyright (c) 1982, 2020, Oracle.  All rights reserved.  
   
Connected to:  
Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
Version 19.9.0.0.0  
   
SQL> alter session set container=p0;  
Session altered.  
SQL> @/home/oracle/ora_kafka_home/app_data/scratch/orakafka_adduser_cluster_KC1_user1.sql  
   
PL/SQL procedure successfully completed.   
Granting permissions on "KC1_CONF_DIR" to "BOOKS_ADMIN"  
Grant succeeded.`

`The above information is written to /home/oracle/ora_kafka_home/app_data/logs/orakafka_adduser_cluster_KC1_user1.log  
Disconnected from Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
Version 19.9.0.0.0  
[oracle@dbass ~]$

Ora si utilizza l'opzione install di orakafka.sh. Passiamo la directory padre (con l'argomento -p) per i dati di questo utente relativi alle sue attività OSaK sui cluster Kafka a cui è stato aggiunto.

Questo comando genera automaticamente due script DDL (Data Definition Language) per noi come riportato di seguito.

   [oracle@dbass bin]$ ./orakafka.sh install -u books_admin -r /home/oracle/ora_kafka_home/books_admin_user_data_dir  
   
Step1: Creation of filesystem location and default directories  
--------------------------------------------------------------  
Created filesystem location directory at /home/oracle/ora_kafka_home/books_admin_user_data_dir/orakafka_location_dir  
Created filesystem default directory at /home/oracle/ora_kafka_home/books_admin_user_data_dir/orakafka_default_dir  
Step1 succeeded.  
   
Step2: Generate DDL for creation of DB location and default directories  
--------------------------------------------------------------  
Execute the following SQL script while connected as sysdba  
to setup database directories:  
 @/home/oracle/ora_kafka_home/app_data/scratch/setup_db_dirs_user1.sql  
On failure, to cleanup location and default directory setup, please run "./orakafka.sh uninstall -u 'BOOKS_ADMIN'"  
Step2 successfully generated script.  
   
Step3: Install ORA_KAFKA package in "BOOKS_ADMIN" user schema  
--------------------------------------------------------------  
Execute the following script in user schema "BOOKS_ADMIN" to  
install ORA_KAFKA package in the user schema  
 @/home/oracle/ora_kafka_home/app_data/scratch/install_orakafka_user1.sql  
On failure, to cleanup ORA_KAFKA package from user schema, please run "./orakafka.sh uninstall -u 'BOOKS_ADMIN'"  
Step3 successfully generated script.  
   
***********SUMMARY************  
   
TODO tasks:  
   
1. Execute the following SQL while connected as sysdba:  
   @/home/oracle/ora_kafka_home/app_data/scratch/setup_db_dirs_user1.sql  
2. Execute the following SQL in user schema:  
   @/home/oracle/ora_kafka_home/app_data/scratch/install_orakafka_user1.sql  
   
The above information is written to /home/oracle/ora_kafka_home/app_data/logs/install.log.2021.02.20-11.33.09  
   
[oracle@dbass bin]

Il primo task consiste nell'eseguire uno script SQL per registrare le due directory, ovvero /home/oracle/ora_kafka_home/books_admin_user_data_dir/orakafka_location_dir e /home/oracle/ora_kafka_home/books_admin_user_data_dir/orakafka_default_dir
con Oracle DB, come oggetti directory.

Come indicato nell'output, è necessario eseguirlo con privilegi SYSDBA come indicato di seguito.

   [oracle@dbass ~]$ sqlplus / as sysdba  
SQL*Plus: Release 19.0.0.0.0 - Production on Sat Feb 20 11:46:04 2021  
Version 19.9.0.0.0  
Copyright (c) 1982, 2020, Oracle.  All rights reserved.  
Connected to:  
Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
Version 19.9.0.0.0  
   
SQL> alter session set container=p0;  
   
Session altered.  
   
SQL> @/home/oracle/ora_kafka_home/app_data/scratch/setup_db_dirs_user1.sql  
Checking if user exists..  
   
PL/SQL procedure successfully completed.  
Creating location and default directories..  
PL/SQL procedure successfully completed.  
Directory created.  
Directory created.  
Grant succeeded.  
Grant succeeded.  
Creation of location dir "BOOKS_ADMIN_KAFKA_LOC_DIR" and default dir "BOOKS_ADMIN_KAFKA_DEF_DIR" completed.  
Grant of required permissions on "BOOKS_ADMIN_KAFKA_LOC_DIR","BOOKS_ADMIN_KAFKA_DEF_DIR" to "BOOKS_ADMIN" completed.  
The above information is written to /home/oracle/ora_kafka_home/app_data/logs/setup_db_dirs_user1.log  
Disconnected from Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
Version 19.9.0.0.0  
[oracle@dbass ~]$

Prendere nota degli oggetti di directory creati dagli script SQL precedenti, ovvero BOOKS_ADMIN_KAFKA_LOC_DIR e BOOKS_ADMIN_KAFKA_DEF_DIR.

Accedere al secondo task TODO.

Qui viene eseguito un altro script SQL come books_admin (e non sysdba) su p0 PDB. Questo script installerà l'accesso Oracle SQL al package e agli oggetti Kafka nello schema appartenente a books_admin.

   [oracle@dbass ~]$ sqlplus books_admin@p0pdb #p0pdb is tns entry for p0 pdb  
SQL*Plus: Release 19.0.0.0.0 - Production on Sat Feb 20 11:52:33 2021  
Version 19.9.0.0.0  
Copyright (c) 1982, 2020, Oracle.  All rights reserved.  
Enter password:  
Last Successful login time: Fri Feb 19 2021 14:04:59 +00:00  
Connected to:  
Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
Version 19.9.0.0.0  
   
SQL> ALTER SESSION SET CONTAINER=p0;  
Session altered.  
   
SQL> @/home/oracle/ora_kafka_home/app_data/scratch/install_orakafka_user1.sql  
Verifying user schema..  
   
PL/SQL procedure successfully completed.  
   
Verifying that location and default directories are accessible..  
   
PL/SQL procedure successfully completed.  
   
Installing ORA_KAFKA package in user schema..  
.. Creating ORA_KAFKA artifacts  
   
Table created.  
Table created.  
Table created.  
Table created.  
Package created.  
No errors.  
Package created.  
No errors.  
Package body created.  
No errors.  
Package body created.  
No errors.  
The above information is written to /home/oracle/ora_kafka_home/app_data/logs/install_orakafka_user1.log  
Disconnected from Oracle Database 19c Standard Edition 2 Release 19.0.0.0.0 - Production  
Version 19.9.0.0.0  
[oracle@dbass ~]$

Registrare il cluster del servizio di streaming con Oracle Database e creare viste per il flusso

In questo passo viene eseguito il cluster kc1 con Oracle Database e vengono create viste per il flusso StreamForOsak.

Tutte le query SQL per questo passo devono essere eseguite come utente books_admin e in PDB p0.

Registrare il cluster con l'accesso Oracle SQL alla procedura Kafka ORA_KAFKA.REGISTER_CLUSTER.

BEGIN  
ORA_KAFKA.REGISTER_CLUSTER  
('kc1', -- cluster name  
'cell-1.streaming.ap-mumbai-1.oci.oraclecloud.com:9092', -- fill up your bootstrap server here   
'BOOKS_ADMIN_KAFKA_DEF_DIR', -- default directory for external table created in previous steps  
'BOOKS_ADMIN_KAFKA_LOC_DIR', -- this directory object too is created in previous steps  
'KC1_CONF_DIR', --config dir for cluster  
'Registering kc1 for this session'); --description   
END;  
/

Creare una tabella Oracle Database denominata BOOKS, che deve avere lo stesso schema dei messaggi nel nostro argomento Kafka StreamForOsak.

CREATE TABLE BOOKS ( -- reference table. It is empty table. Schema of this table must correspond to Kafka Messages  
id int,  
title varchar2(50),  
author_name varchar2(50),  
author_email varchar2(50),  
publisher varchar2(50)  
);  
/

Creare una vista per il nostro argomento StreamForOsak con la procedura memorizzata della Guida dall'accesso a Oracle SQL al pacchetto Kafka versione CREATE_VIEWS come descritto di seguito.

DECLARE  
 application_id VARCHAR2(128);  
 views_created INTEGER;  
BEGIN  
ORA_KAFKA.CREATE_VIEWS  
('kc1', -- cluster name  
'OsakApp0', -- consumer group name that OSS or Kafka cluster sees.  
'StreamForOsak', -- Kafka topic aka OSS stream name  
'CSV', -- format type  
'BOOKS', -- database reference table  
views_created, -- output  
application_id); --output  
dbms_output.put_line('views created  = ' || views_created);  
dbms_output.put_line('application id = ' || application_id);  
END;  
/

L'accesso Oracle SQL a Kafka supporta due formati per messaggi Kafka, CSV e JSON. Qui si utilizza CSV. Pertanto, un messaggio Kafka di esempio conforme allo schema della tabella BOOKS può essere 101, Famous Book, John Smith, john@smith.com, First Software.


tar xvf sqldeveloper-20.4.0.379.2205-20.4.0-379.2205.noarch.rpm

Il passo precedente creerà una vista per ogni partizione dell'argomento. Poiché l'argomento StreamForOsak contiene tre partizioni, saranno disponibili tre viste. Il nome di ciascuna vista sarà nel formato KV_<CLUSTER_NAME>_<GROUP_NAME>_TOPIC_<NUM_OF_PARTITION>.

Di seguito sono riportate tre viste, una per ogni partizione.

Pertanto, la vista KV_KC1_OSAKAPP0_STREAMFOROSAK_0 è mappata alla partizione 0, la vista KV_KC1_OSAKAPP0_STREAMFOROSAK_2 è mappata alla partizione 1 e, infine, KV_KC1_OSAKAPP0_STREAMFOROSAK_0 è mappata alla partizione 2 dell'argomento StreamForOsak.

Produci dati nel flusso

Produrre un messaggio di test per eseguire il flusso di StreamForOsak utilizzando il pulsante Produci messaggio di test dalla console Web di Oracle Cloud Infrastructure (OCI), come mostrato di seguito.

Quando si fa clic su Produci messaggio di test, viene visualizzata una finestra con un campo per immettere il messaggio di test, come mostrato di seguito.

Inserire il seguente messaggio:

200, Andrew Miller Part 1, MS Brown, mb@example.com, First Software

È inutile dire che abbiamo anche usato l'API di produzione Kafka standard in Java/Python o OCI SDK per il servizio Stream, per generare il messaggio nel flusso di servizi Stream.

Dopo aver fatto clic su Produci nella finestra precedente, il messaggio viene pubblicato nel flusso StreamForOsak. È possibile utilizzare la utility Load Message per visualizzare la partizione in cui si trova il sito del messaggio.

Come mostrato in precedenza, il nostro messaggio è stato suddiviso nella partizione 2 dell'argomento StreamForOsak.

Recuperare i messaggi del servizio di streaming utilizzando le viste del database impostate da Oracle SQL accesso alle procedure memorizzate Kafka

È possibile eseguire una query SQL semplice:


SELECT * FROM KV_KC1_OSAKAPP0_STREAMFOROSAK_2_

Oppure, come mostrato di seguito, possiamo semplicemente aprire lo sviluppatore SQL della vista per visualizzare i dati nella vista KV_KC1_OSAKAPP0_STREAMFOROSAK_2:

È stata scelta la vista KV_KC1_OSAKAPP0_STREAMFOROSAK_2, in quanto corrisponde alla partizione 2 del flusso StreamForOsak. Ogni volta che si esegue la query SELECT SQL per questa vista, viene eseguita la partizione corrispondente del relativo argomento associato e vengono recuperati i nuovi messaggi non sottoposti a commit (per il gruppo di consumer OsakApp0) come righe nella vista. Ogni messaggio recupera la propria riga nella vista. Tenere presente che i dati di queste viste non sono persistenti.

Gli occhi curiosi possono inoltre osservare che l'accesso Oracle SQL a Kafka recupera e memorizza informazioni sui metadati come l'offset del messaggio, la relativa partizione e l'indicatore orario in colonne aggiuntive.

Molto probabilmente, si desidera eseguire una nuova query su questa vista e, per ogni esecuzione della query, si desidera che la vista venga spostata in sequenza nel flusso. Puoi ottenere questo risultato con il seguente snippet di codice canonico:

LOOP
ORA_KAFKA.NEXT_OFFSET(‘_KV_KC1_OSAKAPP0_STREAMFOROSAK_2_’);

SELECT * FROM KV_KC1_OSAKAPP0_STREAMFOROSAK_2_;

ORA_KAFKA.UPDATE_OFFSET(‘_KV_KC1_OSAKAPP0_STREAMFOROSAK_2_’);

COMMIT;
END LOOP;

Come detto in precedenza, quando si utilizza l'accesso Oracle SQL a Kafka, gli offset di flusso vengono gestiti da Oracle Database e non dal servizio Stream. Sono disponibili in tabelle di sistema che consentono di tenere traccia del posizionamento degli offset per tutte le partizioni a cui si accede dalla vista KV_KC1_OSAKAPP0_STREAMFOROSAK_2.

La chiamata NEXT_OFFSET associa semplicemente la vista KV_KC1_OSAKAPP0_STREAMFOROSAK_2 a un nuovo set di offset che rappresentano i nuovi dati che risiedono nella partizione Oracle Cloud Infrastructure Streaming a cui si accede dalla vista. UPDATE_OFFSET comunica il numero di righe lette per ciascuna partizione e fa avanzare gli offset in nuove posizioni.

COMMIT garantisce che questa unità di lavoro sia conforme ad ACID.

Conclusione

In questa esercitazione ci siamo concentrati su come installare l'accesso Oracle SQL a Kafka e utilizzarlo con il servizio Stream. Poiché il servizio Stream è compatibile con Kafka, è proprio come un cluster Kafka per l'accesso Oracle SQL a Kafka.

In questa sezione abbiamo appena grattugiato la superficie per quanto riguarda le funzionalità offerte dall'accesso Oracle SQL a Kafka. L'accesso Oracle SQL a Kafka è altamente configurabile. A differenza delle viste, è anche possibile memorizzare i dati di streaming nelle tabelle di Oracle Database, resi persistenti nel disco. Per ulteriori dettagli, fare riferimento ai file README forniti nell'accesso a Oracle SQL all'installazione Kafka e ai riferimenti riportati di seguito.

Riconoscimenti

Autore - Mayur Raleraskar, Solutions Architect

```

Altre risorse di apprendimento

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

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

Titolo e informazioni sul copyright

Stream analytics using Oracle Cloud Infrastructure Streaming and Oracle Database

F50457-01

November 2021