Impostare il flusso di dati per elaborare i log on premise in Oracle Cloud Infrastructure

Introduzione

In questa esercitazione viene eseguito il provisioning dell'infrastruttura cloud e la distribuzione del codice PySpark per il caricamento, l'elaborazione e la memorizzazione automatici dei dati di log on premise nel cloud. I dati vengono posizionati nell'area intermedia dello storage degli oggetti, elaborati in Data Flow e memorizzati in Autonomous Data Warehouse per l'analisi.

Il diagramma seguente mostra l'architettura.

Descrizione dell'illustrazione Architecture-analyze-logs.png

Creare l'applicazione PySpark, quindi eseguire il provisioning di Oracle Cloud Infrastructure (OCI) utilizzando uno script Terraform progettato per questo scopo. Scaricare lo script dal repository GitHub di Oracle e apportare modifiche di minore entità per aggiornare alcuni valori di variabile. Successivamente, applicherai lo script Terraform.

Obiettivi

Una volta completata questa esercitazione, sarà disponibile un sistema operativo che ingloba i dati dallo storage degli oggetti, li elabora nel flusso di dati utilizzando un'applicazione PySpark e memorizza i risultati in Autonomous Data Warehouse.

L'applicazione PySpark fornita qui mostra come autenticare e proteggere i carichi di lavoro e come renderla pronta per l'inserimento nel database. Sarà necessario personalizzarlo per soddisfare le esigenze specifiche di elaborazione dei dati.

Prerequisiti

Prima di avviare questa esercitazione, è necessario completare le applicazioni Connetti flusso di dati PySpark ad Autonomous Database in Oracle Cloud Infrastructure. Questa esercitazione descrive come creare un archivio di dipendenze contenente le librerie Python e Java necessarie per questa esercitazione. In particolare, è necessario il file archive.zip di tale esercitazione.

Per completare questa esercitazione, è necessario disporre dell'accesso a una tenancy su Oracle Cloud Infrastructure con le autorizzazioni per creare gli artifact visualizzati nel diagramma dell'architettura.

In questa esercitazione viene scaricato il codice Terraform che crea gli artifact, come descritto nel diagramma dell'architettura. Il codice si trova in un repository su GitHub. È possibile eseguire il codice Terraform sul computer di sviluppo locale oppure eseguirlo da Cloud Shell in Oracle Cloud Infrastructure. Cloud Shell dispone di Terraform e Git installati.

Se si utilizza il computer locale, vengono applicati i prerequisiti riportati di seguito.

Git installato nel computer di sviluppo. Si tratta della clonazione del repository di codici Terraform.
Terraform installato nel computer di sviluppo. Si tratta della creazione dell'infrastruttura nell'infrastruttura OCI.
Una connessione di rete alla tenancy OCI.

Dettagli connessione ADW

Per connettersi a un database ADW, è necessario passare in un wallet come parte dei parametri di connessione. Il wallet contiene le credenziali e altri dettagli che consentono al portatore di connettersi al database. Il nome file del wallet si trova in genere nel formato wallet_<database-name>.zip, ma potrebbe avere un nome diverso. Il wallet offre una connessione al database, ma le autorizzazioni del database sono determinate dal nome utente e dalla password del database.

Terraform abbinato a questa esercitazione crea un database Autonomous Data Warehouse Sempre gratis denominato log. Memorizza il wallet per il database in due posizioni. Uno si trova all'interno di un bucket nello storage degli oggetti denominato Wallet e l'altro si trova nella directory del computer in cui è stato eseguito il codice Terraform. In entrambi i casi, il nome del file wallet è wallet_logs.zip.

Terraform genera la password ADMIN quando crea il database e la visualizza al termine dell'operazione.

Preparare l'ambiente Oracle Cloud Infrastructure

Prima di creare l'applicazione PySpark ed eseguire Terraform, è necessario eseguire alcune operazioni di configurazione preliminari della tenancy OCI.

È necessario creare un compartimento e un gruppo di utenti dedicato all'applicazione Data Flow. Ciò consente di isolare il flusso di dati da altri utenti nella tenancy.

Terraform può creare il compartimento e i gruppi, ma in questa esercitazione verranno creati personalmente. Ciò consente di eseguire la navigazione e l'utilizzo della console OCI.

Creare un compartimento

Creare un compartimento utilizzato unicamente per contenere artifact, configurazione e altre risorse correlate alla pipeline.

Eseguire il login alla console di Oracle Cloud Infrastructure come utente con privilegi di amministratore.
Nel menu di navigazione della console selezionare Identità, quindi fare clic su Compartimento.
Fare clic su Crea compartimento.
Immettere Dataflow come nome del nuovo compartimento, immettere una descrizione appropriata e assicurarsi che Compartimento padre sia il compartimento radice.
Fare clic su Crea compartimento.
Dopo aver creato il compartimento, creare un record del relativo OCID. Quando si modifica il file terraform.tfvars, sarà necessario disporre dell'OCID successivo.

Crea gruppi

È necessario creare due gruppi. Un gruppo contiene utenti che possono gestire il flusso di dati e l'altro contiene utenti che possono utilizzare il flusso di dati ma non sono autorizzati a gestirlo.

Creare un gruppo.
1. Nel menu di navigazione della console selezionare Identità e fare clic su Gruppi.
2. Fare clic su Crea gruppo.
3. Immettere Dataflow_Admin_Group come nome del gruppo, immettere una descrizione appropriata, quindi fare clic su Crea.
Ripetere il passo precedente per creare un altro gruppo denominato Dataflow_User_Group.
Aggiungere l'utente a Dataflow_Admin_Group. Fare clic su Dataflow_Admin_Group, quindi su Aggiungi utente a gruppo e selezionare l'utente dall'elenco.

Registra lo spazio di nomi dello storage degli oggetti

Fare clic sull'icona del profilo utente e nel menu visualizzato fare clic sul nome della tenancy.
Nella pagina che viene aperta, individuare il valore dello spazio di nomi di storage degli oggetti e crearne una copia per utilizzarlo in un secondo momento.

Genera chiavi API

Terraform utilizza l'API OCI per creare e gestire l'infrastruttura in OCI. A tale scopo, deve disporre della chiave pubblica e dell'impronta digitale di una coppia di chiavi API.

Per generare una coppia di chiavi pubblica/privata:

Fare clic sull'icona del profilo utente e nel menu visualizzato fare clic su Impostazioni utente.
Nella pagina Dettagli utente cercare nella sezione Risorse e fare clic su Chiavi API.
Fare clic su Aggiungi chiave API.
Nel pannello Aggiungi chiave API visualizzato fare clic su Scarica chiave privata.
Nell'anteprima del file di configurazione, copiare il contenuto della casella di testo e salvarlo in una posizione comoda. Queste informazioni saranno necessarie in seguito durante l'impostazione di Terraform.
Fare clic su Chiudi.
Rinominare il file della chiave privata scaricato. Utilizzare un nome semplice come oci_api_key.pem.

Aggiungere la password di amministrazione del database al vault

È necessario creare un segreto vault per la password ADMIN del database e registrare l'OCID. Ciò è necessario per la costante PASSWORD_SECRET_OCID nell'applicazione PySpark creata in un secondo momento.

Per il momento, immettere una password fittizia nel segreto vault. Questo perché non si conosce ancora la password. La password non verrà conosciuta fino al completamento di Terraform. Tuttavia, hai bisogno dell'OCID del segreto per l'applicazione PySpark che Terraform carica nello storage degli oggetti. In questo modo, prima si crea un segreto, si registra l'OCID, quindi dopo il completamento di Terraform, si aggiorna il segreto con la password reale.

Andare a Sicurezza e fare clic su Vault.
Nella sezione Ambito lista, assicurarsi di essere nel compartimento del flusso di dati.
Fare clic su Create Vault.
Nel pannello che viene aperto, immettere Flusso di dati nel campo Nome.
Fare clic su Create Vault.
Quando lo stato del vault diventa Attivo, fare clic su Flusso di dati per aprire la pagina Dettagli vault.
Nella sezione Cifratura principale fare clic su Crea chiave.
Nel pannello Crea chiave immettere Flusso di dati nel campo Nome.
Fare clic su Crea chiave.
Nella sezione Risorse fare clic su Secret.
Fare clic su Crea segreto.
Nella finestra di dialogo Crea segreto, scegliere un compartimento dalla lista Crea nel compartimento.
Nel campo Nome immettere un nome per identificare il segreto. Evitare di fornire informazioni riservate.
Immettere una breve descrizione del segreto per agevolarne l'identificazione nel campo Descrizione. Evitare di fornire informazioni riservate.
Nel campo Chiave di cifratura selezionare Flusso di dati.
Nel campo Modello tipo segreto selezionare Testo normale.
Nel campo Sommario segreto immettere una password fittizia.
Fare clic su Crea segreto.
Quando il pannello viene chiuso, fare clic sul segreto del flusso di dati per aprire la pagina dei dettagli e copiare l'OCID.

Creare l'applicazione PySpark

L'applicazione PySpark crea una sessione Spark che legge i dati di log dallo storage degli oggetti, li trasforma in un dataframe e quindi memorizza il dataframe in una tabella in ADW.

Prima di creare la sessione Spark, è necessario assicurarsi che alcuni moduli importanti vengano importati e impostare alcune costanti da utilizzare in un secondo momento nell'app.

Utilizzare le istruzioni riportate di seguito per importare i moduli rilevanti.

import os
import oci
import base64
import zipfile
from urllib.parse import urlparse

from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_extract

Queste sono le costanti da impostare nella funzione main().

Assicurarsi di inserire i propri valori per OBJECT_STORAGE_NAMESPACE e PASSWORD_SECRET_OCID.

OBJECT_STORAGE_NAMESPACE = "YOUR-OBJECT-STORAGE-NAMESPACE"
INPUT_PATH = "oci://data@{}/sample_logs.log".format(OBJECT_STORAGE_NAMESPACE)
DATABASE_NAME = "logs"
PASSWORD_SECRET_OCID = "ocid1.vaultsecret..... "
WALLET_PATH = "oci://Wallet@{}/wallet_{}.zip".format(OBJECT_STORAGE_NAMESPACE,DATABASE_NAME)
TNS_NAME = "{}_high".format(DATABASE_NAME)
USER="ADMIN"
TARGET_DATABASE_TABLE="processed_logs"

Impostare la sessione Spark e caricare i dati dallo storage degli oggetti. I dati hanno la forma di un file di testo piatto, che viene caricato in un dataframe Spark.

spark_session = SparkSession.builder.appName("Dataflow").getOrCreate()
input_df = spark_session.read.text(INPUT_PATH)
input_df.show(5, truncate=False) # Some output for the Data Flow log file

La dataframe deve essere simile al seguente campione:

+----------------------------------------------------------------------------------+
|value                                                                             |
+----------------------------------------------------------------------------------+
|10.0.0.1 - user1 [10/Mar/2021:13:55:36 -0700] "GET /index.html HTTP/1.0" 200 2326 |
|10.0.0.2 - user2 [10/Mar/2021:14:55:36 -0700] "GET /file1.html HTTP/1.0" 200 9889 |
|10.0.0.3 - user3 [10/Mar/2021:14:55:37 -0700] "GET /file2.html HTTP/1.0" 200 4242 |
|10.0.0.4 - user4 [10/Mar/2021:14:56:36 -0700] "GET /file3.html HTTP/1.0" 200 10267|
|10.0.0.1 - user1 [10/Mar/2021:15:05:36 -0700] "GET /file4.html HTTP/1.0" 200 15678|
+----------------------------------------------------------------------------------+

Come si può vedere, i dati sono costituiti da una singola colonna denominata value che contiene ogni voce di log come singola stringa. Prima di poter trasferire i dati al database, il dataframe deve avere diverse colonne, una per ogni campo nel file di log.

Suddividere il dataframe in colonne.

Nell'esempio seguente, il campo user-identifier (di solito "-") non è incluso nel nuovo dataframe.

hostname = r'(^\S+\.[\S+\.]+\S+)\s'
user = r'\s+.+\s+(.+)\s+\['
timestamp = r'\[(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} -\d{4})]'
method_uri_protocol = r'\"(\S+)\s(\S+)\s*(\S*)\"'
status = r'\s(\d{3})\s'
content_size = r'\s(\d+)$'

logdata_df = input_df.select(
    regexp_extract('value', hostname, 1).alias('hostname'),
    regexp_extract('value', user, 1).alias('user'),
    regexp_extract('value', timestamp, 1).alias('timestamp'),
    regexp_extract('value', method_uri_protocol, 1).alias('method'),
    regexp_extract('value', method_uri_protocol, 2).alias('URI'),
    regexp_extract('value', method_uri_protocol, 3).alias('protocol'),
    regexp_extract('value', status, 1).cast('integer').alias('status'),
    regexp_extract('value', content_size, 1).cast('integer').alias('content_size'))

Questo è anche il luogo in cui puoi applicare uno o più modelli di machine learning. Ad esempio, è possibile eseguire un carico di lavoro di rilevamento delle intrusioni prima di memorizzare i log nel database.

Recuperare le credenziali per il database Autonomous Data Warehouse.

Il codice Terraform abbinato a questa esercitazione crea Autonomus Data Warehouse e memorizza il file wallet nello storage degli oggetti. L'applicazione PySpark deve recuperare il wallet e utilizzare le relative informazioni per la connessione a ADW.

# Get an Object Store client
token_key = "spark.hadoop.fs.oci.client.auth.delegationTokenPath"
token_path = spark_session.sparkContext.getConf().get(token_key)
with open(token_path) as fd:
    delegation_token = fd.read()
signer = oci.auth.signers.InstancePrincipalsDelegationTokenSigner(
    delegation_token=delegation_token)
object_store_client = oci.object_storage.ObjectStorageClient(config={}, signer=signer)

# Extract the wallet file location info
split_url = urlparse(WALLET_PATH)
bucket_name, namespace = split_url.netloc.split("@")
file_name = split_url.path[1:]

# Get the wallet from Object Storage.
# The response contains the wallet and some metadata
response = object_store_client.get_object(namespace, bucket_name, file_name)

# Extract the wallet from response and store it in the Spark work-dir
wallet_path = "/opt/spark/work-dir/"
zip_file_path = os.path.join(wallet_path, "temp.zip")
with open(zip_file_path, "wb") as fd:
    for chunk in response.data.raw.stream(1024 * 1024, decode_content=False):
        fd.write(chunk)
with zipfile.ZipFile(zip_file_path, "r") as zip_ref:
    zip_ref.extractall(wallet_path)

# Extract the wallet contents and add the files to the Spark context
contents = "cwallet.sso ewallet.p12 keystore.jks ojdbc.properties sqlnet.ora tnsnames.ora truststore.jks".split()
for file in contents:
    spark.sparkContext.addFile(os.path.join(wallet_path, file))

Memorizzare i dati nel database Autonomous Data Warehouse.

Lo snippet riportato di seguito mostra come connettersi a ADW e scrivere il dataframe nella tabella di destinazione. Se la tabella non esiste, viene creata. Se la tabella esiste, viene eliminata e ricreata a causa dell'opzione "overwrite" nel parametro mode.

La password viene recuperata da Oracle Cloud Infrastructure Vault. La password viene posizionata nel vault dopo aver eseguito il codice Terraform. L'inserimento della password nel Vault viene eseguito nei passi successivi all'esecuzione di Terraform.

adw_url = "jdbc:oracle:thin:@{}?TNS_ADMIN={}".format(TNS_NAME, wallet_path)

# Retrieve the database password for user 'ADMIN' from OCI Vault
# The password is stored as base64 text, so it must be decoded
with open(token_path) as fd:
    delegation_token = fd.read()
signer = oci.auth.signers.InstancePrincipalsDelegationTokenSigner(
    delegation_token=delegation_token)
secrets_client = oci.secrets.SecretsClient(config={}, signer=signer)
response = secrets_client.get_secret_bundle(PASSWORD_SECRET_OCID)

base64_password = response.data.secret_bundle_content.content
base64_secret_bytes = base64_password.encode("ascii")
base64_message_bytes = base64.b64decode(base64_secret_bytes)
password = base64_message_bytes.decode("ascii")

# Set up some properties for connecting to the database
properties = {
    "driver": "oracle.jdbc.driver.OracleDriver",
    "oracle.net.tns_admin": TNS_NAME,
    "password": password,
    "user": USER,
}

# Write the dataframe to the database
adw_url = "jdbc:oracle:thin:@{}?TNS_ADMIN={}".format(TNS_NAME, wallet_path)
logdata_df.write.jdbc(url=adw_url, table=TARGET_DATABASE_TABLE, 
    mode="Overwrite", properties=properties)

Salvare l'applicazione PySpark con il nome dataflow-app.py.

Scaricare e applicare il codice Terraform

Il codice Terraform è disponibile nel repository quickstart di Oracle su GitHub. Il codice esegue il provisioning dei componenti visualizzati nel diagramma dell'architettura.

Per scaricare e applicare il codice:

Passare a una directory comoda e duplicare il repository.

git clone https://github.com/oracle-quickstart/oci-arch-dataflow-store-analyze-data
Copiare il file di dipendenza archive.zip nella directory oci-arch-dataflow-store-analyze-data.

Nota: se non si dispone del file archive.zip, vedere la sezione Prerequisiti per scoprire come ottenerlo.
Spostarsi nella directory oci-arch-dataflow-store-analyze-data.

cd oci-arch-dataflow-store-analyze-data
Individuare il file dell'applicazione PySpark dataflow-app.py, che è un file stub, e sostituirlo con quello scritto.
Aprire il file terraform.tfvars in un editor di testo.
Aggiornare il valore per object_storage_namespace in terraform.tfvars. Utilizzare il valore registrato in precedenza.
Modificare env-vars.bat o env-vars.sh in base al sistema operativo del computer in uso. Questi script vengono utilizzati per impostare le variabili di ambiente utilizzate da Terraform.

In Windows il file env-vars.bat deve essere simile al seguente esempio. Accertarsi di incollare i valori per la propria tenancy.
```
@echo off
set TF_VAR_tenancy_ocid=ocid1.tenancy.oc1..
set TF_VAR_user_ocid=ocid1.user.oc1..
set TF_VAR_compartment_ocid=ocid1.compartment.oc1..
set TF_VAR_private_key_path=%HOMEPATH%\.ssh\oci_api_key.pem
set TF_VAR_fingerprint=2a:b4:ef:9c:f1...
set TF_VAR_region=ca-toronto-1
```
Su Linux e Mac, il file env-vars.sh deve essere simile al seguente esempio. Accertarsi di incollare i valori per la propria tenancy.
```
export TF_VAR_tenancy_ocid=ocid1.tenancy.oc1..
export TF_VAR_user_ocid=ocid1.user.oc1..
export TF_VAR_compartment_ocid=ocid1.compartment.oc1..
export TF_VAR_private_key_path=~/.ssh/oci_api_key.pem
export TF_VAR_fingerprint=2a:b4:ef:9c:f1...
export TF_VAR_region=ca-toronto-1
```
Dopo aver aggiornato i file con le proprie informazioni sulla tenancy e sull'utente, accertarsi che le variabili di ambiente siano impostate.

In Windows, eseguire il comando seguente:
```
env-vars.bat
```
In Linux o Mac, eseguire il comando seguente:
```
source env-vars.sh
```
In un terminale, eseguire il comando init

terraform init
Esegui il provisioning delle risorse in Oracle Infrastructure Cloud.

terraform apply

Al termine, l'applicazione Terraform visualizza la password ADMIN per il database creato.
Copiare la password visualizzata dall'applicazione Terraform.
Verificare che il provisioning delle risorse sia stato eseguito.
1. Andare alla console OCI e aprire il menu di navigazione.
2. Scorrere fino alla sezione Gestione del controllo e amministrazione, espandere Gestione del controllo, quindi fare clic su Esplora risorse.
3. Selezionare il compartimento Dataflow creato in precedenza.
Dopo alcuni secondi, le risorse vengono visualizzate in una tabella che è possibile filtrare e ordinare.

Aggiornare la password ADMIN nel vault

In precedenza, era stata aggiunta una password fittizia al vault OCI per ottenere l'OCID relativo da utilizzare nell'applicazione PySpark eseguita nel flusso di dati. Ora è necessario aggiornare la password al valore reale. Il valore reale è stato visualizzato nel terminale al termine del comando di applicazione Terraform.

Nel menu di navigazione della console selezionare Sicurezza e fare clic su Vault.
Fare clic sul nome del vault creato in precedenza.
Nella pagina Dettagli vault visualizzata, cercare nella sezione Risorse e fare clic su Secret.
Fare clic sul nome del segreto creato in precedenza.
Fare clic su Crea versione segreta.
Assicurarsi che Modello tipo segreto sia impostato su Testo filtro e incollare la password ADMIN del database nella casella Sommario segreto.
Fare clic su Crea versione segreta.

Eseguire l'applicazione Data Flow

È ora il momento di eseguire l'applicazione PySpark.

Nel menu di navigazione della console selezionare Flusso di dati e fare clic su Applicazioni.
Fare clic su Analizza log in locale per aprire la pagina Dettagli applicazione.
Fare clic su Run per aprire il pannello Run Python Application.
Fare clic su Esegui.

Viene visualizzata la pagina Esecuzioni flusso di dati in cui è possibile monitorare l'avanzamento dell'esecuzione. L'allocazione delle risorse richiede alcuni minuti e l'attivazione di tutto. Al termine dell'esecuzione, fare clic sul nome dell'esecuzione per aprire la pagina Dettagli esecuzione.

Nella pagina Dettagli esecuzione è possibile scaricare ed esaminare i log prodotti. Se l'esecuzione non riesce per qualsiasi motivo, è possibile esaminare i log per determinare l'origine del problema.

Riconoscimenti

Autore - Jeff Schering (User Assistance Developer)
Contributori - Prashant Jha (Director Product Management, OCI Big Data Development)

Per saperne di più

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.

Altre risorse di apprendimento

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

Titolo e informazioni sul copyright

Store and analyze your on-premises logs in Oracle Cloud Infrastructure

F52095-01

January 2022