외부 메타스토어로 IBM Cloud Databases for PostgreSQL 사용 

IBM Cloud Databases for PostgreSQL 을 사용하여 IBM Analytics Engine Spark 클러스터 외부에서 메타데이터를 구체화할 수 있습니다.

  1. IBM Cloud Databases for PostgreSQL 인스턴스를 작성하십시오.  Databases for PostgreSQL 의 내용을 참조하십시오.

    요구사항에 따라 구성을 선택하십시오. 엔드포인트 구성에 대해 공용 및 사설 네트워크 둘 다 를 선택해야 합니다. 인스턴스 및 서비스 인스턴스 신임 정보를 작성한 후 데이터베이스 이름, 포트, 사용자 이름, 비밀번호 및 인증서를 기록해 두십시오.

  2. 애플리케이션 코드를 유지보수하는 IBM Cloud Object Storage 버킷에 Databases for PostgreSQL 인증서를 업로드하십시오.

    Databases for PostgreSQL에 액세스하려면 클라이언트 인증서를 제공해야 합니다. Databases for PostgreSQL 인스턴스의 서비스 인증 정보에서 Base64 디코딩된 인증서를 가져오고 파일 (이름은 postgres.cert) 을 특정 IBM Cloud 위치의 Object Storage 버킷에 업로드하십시오. 나중에 이 인증서를 다운로드하여 메타스토어에 연결하기 위해 IBM Analytics Engine 인스턴스 Spark 워크로드에서 사용할 수 있도록 해야 합니다.

  3. Databases for PostgreSQL 인증서를 포함하도록 IBM Analytics Engine 인스턴스를 사용자 정의하십시오. 스크립트 기반 사용자 정의 를 참조하십시오.

    이 단계에서는 IBM Analytics Engine 인스턴스를 사용자 정의하여 라이브러리 세트를 통해 인스턴스에 대해 실행되는 모든 Spark 워크로드에서 Databases for PostgreSQL 인증서를 사용할 수 있도록 합니다.

    1. 스크립트 기반 사용자 정의 의 페이지에서 IBM Cloud Object Storage 버킷으로 customization_script.py 를 업로드하십시오.

    2. spark-submit REST API를 사용하는 postgres-cert-customization-submit.json 를 실행하여 인스턴스를 사용자 정의하십시오. 참고로, 코드는 IBM Cloud Object Storage에 업로드한 postgres.cert 를 참조합니다.

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

      라이브러리 세트 이름 certificate_library_set 은 사용자가 지정한 Databases for PostgreSQL 메타스토어 연결 매개변수 ae.spark.librarysets 의 값과 일치해야 합니다.

  4. Spark 애플리케이션 페이로드의 일부 또는 인스턴스 기본값으로 다음 Databases for PostgreSQL 메타스토어 연결 매개변수를 지정하십시오. 아래의 "spark.hadoop.javax.jdo.option.ConnectionURL" 매개변수에 대해 사설 엔드포인트를 사용하는지 확인하십시오.

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

  5. 인스턴스를 작성할 때 Databases for PostgreSQL 데이터베이스의 공용 스키마에 테이블이 없으므로 Databases for PostgreSQL 인스턴스에서 Hive 메타스토어 스키마를 설정하십시오. 이 단계는 메타데이터를 저장할 수 있도록 Hive 스키마 관련 DDL을 실행합니다. postgres-create-schema.py 라는 다음 Spark 애플리케이션을 실행하면 인스턴스의 "공용" 스키마에 대해 작성된 Hive 메타데이터 테이블이 표시됩니다.

    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()

  6. 이제 postgres-parquet-table-create.py 라는 다음 스크립트를 실행하여 Databases for PostgreSQL 데이터베이스에 있는 IBM Cloud Object Storage 의 메타데이터로 Parquet 테이블을 작성하십시오.

    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()

  7. postgres-parquet-table-select.py 라는 다음 PySpark 스크립트를 실행하여 다른 Spark 워크로드의 메타데이터로 이 Parquet 테이블에 액세스하십시오.

    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()