Utilización de IBM Cloud Databases for PostgreSQL como almacén de metadatos externo

Puede utilizar IBM Cloud Databases for PostgreSQL para externalizar metadatos fuera del clúster de Spark IBM Analytics Engine.

Cree una instancia de IBM Cloud Databases for PostgreSQL. Consulte Databases for PostgreSQL.

Elija las configuraciones en función de sus requisitos. Asegúrese de elegir Red pública y privada para la configuración de punto final. Después de haber creado la instancia y las credenciales de la instancia de servicio, anote el nombre de la base de datos, el puerto, el nombre de usuario, la contraseña y el certificado.
Cargue el certificado Databases for PostgreSQL en un grupo IBM Cloud Object Storage en el que está manteniendo el código de aplicación.

Para acceder a Databases for PostgreSQL, debe proporcionar un certificado de cliente. Obtenga el certificado descodificado Base64 de las credenciales de servicio de la instancia de Databases for PostgreSQL y cargue el archivo (nombre que indica, postgres.cert) a un grupo Object Storage en una ubicación específica de IBM Cloud. Más adelante tendrá que descargar este certificado y hacerlo disponible en las cargas de trabajo Spark de la instancia de IBM Analytics Engine para conectarse al almacén de metadatos
Personalice la instancia de IBM Analytics Engine para incluir el certificado Databases for PostgreSQL. Consulte Personalización basada en script.

Este paso personaliza la instancia de IBM Analytics Engine para que el certificado Databases for PostgreSQL esté disponible para todas las cargas de trabajo de Spark que se ejecutan en la instancia a través del conjunto de bibliotecas.
1. Cargue customization_script.py desde la página en Personalización basada en script en un grupo IBM Cloud Object Storage.
2. Ejecute postgres-cert-customization-submit.json que utiliza la API REST spark-submit para personalizar la instancia. Tenga en cuenta que el código hace referencia a postgres.cert que ha subido a 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"
    }
} 
```
  Tenga en cuenta que el nombre de conjunto de bibliotecas certificate_library_set debe coincidir con el valor del parámetro de conexión de almacén de metadatos Databases for PostgreSQL ae.spark.librarysets que ha especificado.

Especifique los siguientes parámetros de conexión de almacén de metadatos Databases for PostgreSQL como parte de la carga útil de la aplicación Spark o como valores predeterminados de instancia. Asegúrese de utilizar el punto final privado para el parámetro "spark.hadoop.javax.jdo.option.ConnectionURL" siguiente:

"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"

Configure el esquema de almacén de metadatos Hive en la instancia de Databases for PostgreSQL porque no hay tablas en el esquema público de la base de datos Databases for PostgreSQL al crear la instancia. Este paso ejecuta el DDL relacionado con el esquema Hive para que los datos del almacén de metadatos se puedan almacenar en ellos. Después de ejecutar la siguiente aplicación Spark denominada postgres-create-schema.py, verá las tablas de metadatos de Hive creadas en el esquema "público" de la instancia.
```
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()
```

Ahora ejecute el siguiente script llamado postgres-parquet-table-create.py para crear una tabla Parquet con metadatos de la base de datos IBM Cloud Object Storage en la base de datos 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()

Ejecute el siguiente script PySpark denominado postgres-parquet-table-select.py para acceder a esta tabla Parquet con metadatos de otra carga de trabajo de 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()