Utilizzo di IBM Cloud Databases for PostgreSQL come metastore esterno

Puoi utilizzare IBM Cloud Databases for PostgreSQL per esternalizzare i metadati al di fuori del cluster Spark IBM Analytics Engine.

Crea un'istanza IBM Cloud Databases for PostgreSQL. Vedi Databases for PostgreSQL.

Scegliere le configurazioni in base alle esigenze. Assicurati di scegliere Sia la rete pubblica che quella privata per la configurazione dell'endpoint. Dopo aver creato l'istanza e le relative credenziali, prendi nota del nome del database, della porta, del nome utente, della password e del certificato.
Carica il certificato Databases for PostgreSQL in un bucket IBM Cloud Object Storage in cui stai gestendo il codice della tua applicazione.

Per accedere a Databases for PostgreSQL, devi fornire un certificato client. Ottieni il certificato decodificato Base64 dalle credenziali del servizio dell'istanza Databases for PostgreSQL e carica il file (nome: postgres.cert) a un bucket Object Storage in un'ubicazione IBM Cloud specifica. Successivamente dovrai scaricare questo certificato e renderlo disponibile nei carichi di lavoro Spark dell'istanza IBM Analytics Engine per la connessione al metastore
Personalizza l'istanza IBM Analytics Engine per includere il certificato Databases for PostgreSQL. Vedere Personalizzazione basata sugli script.

Questo passo personalizza l'istanza IBM Analytics Engine per rendere disponibile il certificato Databases for PostgreSQL per tutti i carichi di lavoro Spark eseguiti sull'istanza tramite la serie di librerie.
1. Carica il customization_script.py dalla pagina in Personalizzazione basata sullo script in un bucket IBM Cloud Object Storage.
2. Esegui postgres-cert-customization-submit.json che utilizza l'API REST spark - submit per personalizzare l'istanza. Tieni presente che il codice fa riferimento a postgres.cert che hai caricato in IBM Cloud Object Storage.
```
{
    "application_details": 
    {
       "application": "/opt/ibm/customization-scripts/customize_instance_app.py",
    "arguments": ["{\"library_set\":{\"action\":\"add\",\"name\":\"certificate_library_set\",\"script\":{\"source\":\"py_files\",\"params\":[\"https://s3.direct.<CHANGME>.cloud-object-storage.appdomain.cloud\",\"<CHANGEME_BUCKET_NAME>\",\"postgres.cert\",\"<CHANGEME_ACCESS_KEY>\",\"<CHANGEME_SECRET_KEY>\"]}}}"],
    "py-files": "cos://CHANGEME_BUCKET_NAME.mycosservice/customization_script.py"
    }
} 
```
  Tieni presente che il nome della serie di librerie certificate_library_set deve corrispondere al valore del parametro di connessione metastore Databases for PostgreSQL ae.spark.librarysets che hai specificato.

Specifica i seguenti parametri di connessione metastore Databases for PostgreSQL come parte del payload dell'applicazione Spark o come valori predefiniti dell'istanza. Assicurati di utilizzare l'endpoint privato per il parametro "spark.hadoop.javax.jdo.option.ConnectionURL" riportato di seguito:

"spark.hadoop.javax.jdo.option.ConnectionDriverName": "org.postgresql.Driver",
"spark.hadoop.javax.jdo.option.ConnectionUserName": "ibm_cloud_<CHANGEME>",
"spark.hadoop.javax.jdo.option.ConnectionPassword": "<CHANGEME>",
"spark.sql.catalogImplementation": "hive",
"spark.hadoop.hive.metastore.schema.verification": "false",
"spark.hadoop.hive.metastore.schema.verification.record.version": "false",
"spark.hadoop.datanucleus.schema.autoCreateTables":"true",
"spark.hadoop.javax.jdo.option.ConnectionURL": "jdbc:postgresql://<CHANGEME>.databases.appdomain.CHANGEME/ibmclouddb?sslmode=verify-ca&sslrootcert=/home/spark/shared/user-libs/certificate_library_set/custom/postgres.cert&socketTimeout=30",
"ae.spark.librarysets":"certificate_library_set"

Configura lo schema metastore Hive nell'istanza Databases for PostgreSQL perché nello schema pubblico non sono presenti tabelle del database Databases for PostgreSQL quando crei l'istanza. Questo passo esegue la DDL relativa allo schema Hive in modo che i dati metastore possano essere memorizzati in essi. Dopo aver eseguito la seguente applicazione Spark denominata postgres-create-schema.py, vedrai le tabella di metadati Hive create rispetto allo schema "pubblico" dell'istanza.
```
from pyspark.sql import SparkSession
import time
def init_spark():
  spark = SparkSession.builder.appName("postgres-create-schema").getOrCreate()
  sc = spark.sparkContext
  return spark,sc
def create_schema(spark,sc):
  tablesDF=spark.sql("SHOW TABLES")
  tablesDF.show()
  time.sleep(30)
def main():
  spark,sc = init_spark()
  create_schema(spark,sc)
if __name__ == '__main__':
  main()
```

Ora esegui il seguente script denominato postgres-parquet-table-create.py per creare una tabella Parquet con i metadati dal database IBM Cloud Object Storage nel Databases for PostgreSQL.

from pyspark.sql import SparkSession
import time
def init_spark():
  spark = SparkSession.builder.appName("postgres-create-parquet-table-test").getOrCreate()
  sc = spark.sparkContext
  return spark,sc
def generate_and_store_data(spark,sc):
  data =[("1","Romania","Bucharest","81"),("2","France","Paris","78"),("3","Lithuania","Vilnius","60"),("4","Sweden","Stockholm","58"),("5","Switzerland","Bern","51")]
  columns=["Ranking","Country","Capital","BroadBandSpeed"]
  df=spark.createDataFrame(data,columns)
  df.write.parquet("cos://<CHANGEME-BUCKET>.mycosservice/broadbandspeed")
def create_table_from_data(spark,sc):
  spark.sql("CREATE TABLE MYPARQUETBBSPEED (Ranking STRING, Country STRING, Capital STRING, BroadBandSpeed STRING) STORED AS PARQUET  location 'cos://CHANGEME-BUCKET.mycosservice/broadbandspeed/'")
  df2=spark.sql("SELECT * from MYPARQUETBBSPEED")
  df2.show()
def main():
  spark,sc = init_spark()
  generate_and_store_data(spark,sc)
  create_table_from_data(spark,sc)
  time.sleep(30)
if __name__ == '__main__':
  main()

Esegui il seguente script PySpark denominato postgres-parquet-table-select.py per accedere a questa tabella Parquet con metadati da un altro carico di lavoro Spark:

from pyspark.sql import SparkSession
import time
def init_spark():
  spark = SparkSession.builder.appName("postgres-select-parquet-table-test").getOrCreate()
  sc = spark.sparkContext
  return spark,sc
def select_data_from_table(spark,sc):
  df=spark.sql("SELECT * from MYPARQUETBBSPEED")
  df.show()
def main():
  spark,sc = init_spark()
  select_data_from_table(spark,sc)
  time.sleep(60)
if __name__ == '__main__':
 main()