전제조건

• Spark 애플리케이션에서 사용되는 데이터를 저장할 Cloud Object Storage를 만듭니다. Cloud Object Storage와 버킷을 만들려면 스토리지 버킷 만들기 를 참조하세요. 두 개의 버킷을 프로비저닝할 수 있는데, watsonx.data 테이블을 저장하는 데이터 버킷과 Spark 애플리케이션 코드를 유지 관리하는 애플리케이션 버킷이 있습니다.
• Cloud Object Storage 버킷을 watsonx.data에 등록하세요. 자세한 내용은 버킷 카탈로그 쌍 추가하기 를 참조하세요.
• Spark 애플리케이션을 스토리지에 업로드하려면 데이터 업로드하기 를 참조하세요.
• watsonx.data 내에 스키마 또는 테이블을 생성하려면 IAM 관리자 역할 또는 MetastoreAdmin 역할이 있어야 합니다.

프로시저

스파크 액세스 제어 확장 프로그램은 외부 스파크 엔진을 지원합니다.

1. Spark 액세스 제어 확장 기능을 사용하려면 add authz.IBMSparkACExtension to spark.sql.extensions 로 Spark 구성을 업데이트해야 합니다.

2. 다음 Python 애플리케이션을 iceberg.py로 저장합니다.

빙산을 예로 들 수 있습니다. 또한, Hudi 및 Delta Lake Hive 카탈로그를 사용할 수 있습니다.

from pyspark.sql import SparkSession
import os

def init_spark():
    spark = SparkSession.builder \
        .appName("lh-spark-app") \
        .enableHiveSupport() \
        .getOrCreate()
    return spark

def create_database(spark):
    # Create a database in the lakehouse catalog
    spark.sql("create database if not exists lakehouse.demodb LOCATION 's3a://lakehouse-bucket/'")

def list_databases(spark):
    # list the database under lakehouse catalog
    spark.sql("show databases from lakehouse").show()

def basic_iceberg_table_operations(spark):
    # demonstration: Create a basic Iceberg table, insert some data and then query table
    spark.sql("create table if not exists lakehouse.demodb.testTable(id INTEGER, name VARCHAR(10), age INTEGER, salary DECIMAL(10, 2)) using iceberg").show()
    spark.sql("insert into lakehouse.demodb.testTable values(1,'Alan',23,3400.00),(2,'Ben',30,5500.00),(3,'Chen',35,6500.00)")
    spark.sql("select * from lakehouse.demodb.testTable").show()

def create_table_from_parquet_data(spark):
    # load parquet data into dataframe
    df = spark.read.option("header",True).parquet("file:///spark-vol/yellow_tripdata_2022-01.parquet")
    # write the dataframe into an Iceberg table
    df.writeTo("lakehouse.demodb.yellow_taxi_2022").create()
    # describe the table created
    spark.sql('describe table lakehouse.demodb.yellow_taxi_2022').show(25)
    # query the table
    spark.sql('select * from lakehouse.demodb.yellow_taxi_2022').count()

def ingest_from_csv_temp_table(spark):
    # load csv data into a dataframe
    csvDF = spark.read.option("header",True).csv("file:///spark-vol/zipcodes.csv")
    csvDF.createOrReplaceTempView("tempCSVTable")
    # load temporary table into an Iceberg table
    spark.sql('create or replace table lakehouse.demodb.zipcodes using iceberg as select * from tempCSVTable')
    # describe the table created
    spark.sql('describe table lakehouse.demodb.zipcodes').show(25)
    # query the table
    spark.sql('select * from lakehouse.demodb.zipcodes').show()

def ingest_monthly_data(spark):
    df_feb = spark.read.option("header",True).parquet("file:///spark-vol/yellow_tripdata_2022-02.parquet")
    df_march = spark.read.option("header",True).parquet("file:///spark-vol/yellow_tripdata_2022-03.parquet")
    df_april = spark.read.option("header",True).parquet("file:///spark-vol/yellow_tripdata_2022-04.parquet")
    df_may = spark.read.option("header",True).parquet("file:///spark-vol/yellow_tripdata_2022-05.parquet")
    df_june = spark.read.option("header",True).parquet("file:///spark-vol/yellow_tripdata_2022-06.parquet")
    df_q1_q2 = df_feb.union(df_march).union(df_april).union(df_may).union(df_june)
    df_q1_q2.write.insertInto("lakehouse.demodb.yellow_taxi_2022")

def perform_table_maintenance_operations(spark):
    # Query the metadata files table to list underlying data files
    spark.sql("SELECT file_path, file_size_in_bytes FROM lakehouse.demodb.yellow_taxi_2022.files").show()
    # There are many smaller files compact them into files of 200MB each using the
    # `rewrite_data_files` Iceberg Spark procedure
    spark.sql(f"CALL lakehouse.system.rewrite_data_files(table => 'demodb.yellow_taxi_2022', options => map('target-file-size-bytes','209715200'))").show()
    # Again, query the metadata files table to list underlying data files; 6 files are compacted
    # to 3 files
    spark.sql("SELECT file_path, file_size_in_bytes FROM lakehouse.demodb.yellow_taxi_2022.files").show()
    # List all the snapshots
    # Expire earlier snapshots. Only latest one with compacted data is required
    # Again, List all the snapshots to see only 1 left
    spark.sql("SELECT committed_at, snapshot_id, operation FROM lakehouse.demodb.yellow_taxi_2022.snapshots").show()
    #retain only the latest one
    latest_snapshot_committed_at = spark.sql("SELECT committed_at, snapshot_id, operation FROM lakehouse.demodb.yellow_taxi_2022.snapshots").tail(1)[0].committed_at
    print (latest_snapshot_committed_at)
    spark.sql(f"CALL lakehouse.system.expire_snapshots(table => 'demodb.yellow_taxi_2022',older_than => TIMESTAMP '{latest_snapshot_committed_at}',retain_last => 1)").show()
    spark.sql("SELECT committed_at, snapshot_id, operation FROM lakehouse.demodb.yellow_taxi_2022.snapshots").show()
    # Removing Orphan data files
    spark.sql(f"CALL lakehouse.system.remove_orphan_files(table => 'demodb.yellow_taxi_2022')").show(truncate=False)
    # Rewriting Manifest Files
    spark.sql(f"CALL lakehouse.system.rewrite_manifests('demodb.yellow_taxi_2022')").show()

def evolve_schema(spark):
    # demonstration: Schema evolution
    # Add column fare_per_mile to the table
    spark.sql('ALTER TABLE lakehouse.demodb.yellow_taxi_2022 ADD COLUMN(fare_per_mile double)')
    # describe the table
    spark.sql('describe table lakehouse.demodb.yellow_taxi_2022').show(25)

def clean_database(spark):
    # clean-up the demo database
    spark.sql('drop table if exists lakehouse.demodb.testTable purge')
    spark.sql('drop table if exists lakehouse.demodb.zipcodes purge')
    spark.sql('drop table if exists lakehouse.demodb.yellow_taxi_2022 purge')
    spark.sql('drop database if exists lakehouse.demodb cascade')

def main():
    try:
        spark = init_spark()
        create_database(spark)
        list_databases(spark)
        basic_iceberg_table_operations(spark)
        # demonstration: Ingest parquet and csv data into a watsonx.data Iceberg table
        create_table_from_parquet_data(spark)
        ingest_from_csv_temp_table(spark)
        # load data for the month of February to June into the table yellow_taxi_2022 created above
        ingest_monthly_data(spark)
        # demonstration: Table maintenance
        perform_table_maintenance_operations(spark)
        # demonstration: Schema evolution
        evolve_schema(spark)
    finally:
        # clean-up the demo database
        clean_database(spark)
        spark.stop()

if __name__ == '__main__':
    main()

3. Spark 애플리케이션을 제출하려면 매개변수 값을 지정하고 다음 curl 명령을 실행합니다. 다음 예시는 iceberg.py 애플리케이션을 제출하는 명령을 보여줍니다.

curl --request POST   --url https://<region>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications    --header 'Authorization: Bearer <token>'   --header 'Content-Type: application/json'   --header 'Lhinstanceid: <instance_id>'   --data '{
  "application_details": {
  "conf": {
      "spark.hadoop.fs.s3a.bucket.<wxd-data-bucket-name>.endpoint": "<wxd-data-bucket-endpoint>",
      "spark.hadoop.fs.cos.<COS_SERVICE_NAME>.endpoint": "<COS_ENDPOINT>",
      "spark.hadoop.fs.cos.<COS_SERVICE_NAME>.secret.key": "<COS_SECRET_KEY>",
      "spark.hadoop.fs.cos.<COS_SERVICE_NAME>.access.key": "<COS_ACCESS_KEY>"
      "spark.sql.catalogImplementation": "hive",
      "spark.sql.iceberg.vectorization.enabled":"false",
        "spark.sql.catalog.<wxd-bucket-catalog-name>":"org.apache.iceberg.spark.SparkCatalog",
      "spark.sql.catalog.<wxd-bucket-catalog-name>.type":"hive",
      "spark.sql.catalog.<wxd-bucket-catalog-name>.uri":"thrift://<wxd-catalog-metastore-host>",
      "spark.hive.metastore.client.auth.mode":"PLAIN",
      "spark.hive.metastore.client.plain.username":"<username>",
      "spark.hive.metastore.client.plain.password":"xxx",
      "spark.hive.metastore.use.SSL":"true",
      "spark.hive.metastore.truststore.type":"JKS",
      "spark.hive.metastore.truststore.path":"<truststore_path>",
      "spark.hive.metastore.truststore.password":"changeit",
        "spark.hadoop.fs.s3a.bucket.<wxd-data-bucket-name>.aws.credentials.provider":"com.ibm.iae.s3.credentialprovider.WatsonxCredentialsProvider",
        "spark.hadoop.fs.s3a.bucket.<wxd-data-bucket-name>.custom.signers":"WatsonxAWSV4Signer:com.ibm.iae.s3.credentialprovider.WatsonxAWSV4Signer",
        "spark.hadoop.fs.s3a.bucket.<wxd-data-bucket-name>.s3.signing-algorithm":"WatsonxAWSV4Signer",
        "spark.hadoop.wxd.cas.endpoint":"<cas_endpoint>/cas/v1/signature",
        "spark.hadoop.wxd.instanceId":"<instance_crn>",
        "spark.hadoop.wxd.apiKey":"Basic xxx",
        "spark.wxd.api.endpoint":"<wxd-endpoint>",
        "spark.driver.extraClassPath":"opt/ibm/connectors/wxd/spark-authz/cpg-client-1.0-jar-with-dependencies.jar:/opt/ibm/connectors/wxd/spark-authz/ibmsparkacextension_2.12-1.0.jar",
        "spark.sql.extensions":"<required-storage-support-extension>,authz.IBMSparkACExtension"

    },
    "application": "cos://<BUCKET_NAME>.<COS_SERVICE_NAME>/<python_file_name>",
  }
}

버전부터 2.2.0watsonx.data, 와 ibmlhtoken 를 사용자 이름으로 ibmlhapikey 사용하는 인증은 더 이상 권장되지 않습니다. 이러한 형식은 출시 2.3.0 과정에서 단계적으로 폐지됩니다. 향후 버전과의 호환성을 보장하려면 새 형식을 사용하세요:ibmlhapikey_<username> 및 ibmlhtoken_<username>.

매개변수 값:

• <region> 인스턴스가 프로비저닝되는 지역. 예: us-south 지역.
• <spark_engine_id>: Spark 인스턴스의 고유 식별자. ID를 찾는 방법에 대한 정보는 스파크 엔진 세부 사항 관리하기를 참고하세요.
• <token> 서비스 인스턴스에 대한 액세스 토큰을 얻으려면. 토큰 생성에 대한 자세한 내용은 토큰 생성을 참고하세요.
• <instance_id> watsonx.data 클러스터 인스턴스 URL 인스턴스 ID입니다. 예: crn:v1:staging:public:lakehouse:us-south:a/7bb9e380dc0c4bc284592b97d5095d3c:5b602d6a-847a-469d-bece-0a29124588c0::.
• <wxd-data-bucket-name>: 인프라 관리자로부터 스파크 엔진에 연결된 데이터 버킷의 이름입니다.
• <wxd-data-bucket-endpoint>: 위에서 언급한 데이터 버킷에 액세스하기 위한 엔드포인트의 호스트 이름입니다. 예를 들어, us-south 리전 클라우드 오브젝트 스토리지 버킷의 경우 s3.us-south.cloud-object-storage.appdomain.cloud로 설정합니다.
• <wxd-bucket-catalog-name>: 데이터 버킷과 연결된 카탈로그의 이름입니다.
• <wxd-catalog-metastore-host>: 등록된 버킷과 연결된 메타스토어입니다.
• <cos_bucket_endpoint>: 메타스토어 호스트 값을 입력합니다. 자세한 내용은 저장소 세부 정보 를 참조하세요.
• <access_key>: access_key_id를 입력합니다. 자세한 내용은 저장소 세부 정보 를 참조하세요.
• <secret_key>: secret_access_key를 입력합니다. 자세한 내용은 저장소 세부 정보 를 참조하세요.
• <truststore_path>: 트러스토어 인증서를 업로드할 COS 경로를 입력합니다. 예: cos://di-bucket.di-test/1902xx-truststore.jks. 신뢰 저장소 생성에 대한 자세한 내용은 자체 서명 인증서 가져오기 를 참조하세요.
• <cas_endpoint>: DAS(데이터 액세스 서비스) 엔드포인트입니다. DAS 엔드포인트를 가져오려면 DAS 엔드포인트 가져오기 를 참조하세요.
• <username>: watsonx.data 인스턴스의 사용자 이름입니다. 여기, ibmlhapikey.
• <apikey> : base64 인코딩된 `ibmlhapikey_<user_id>:<IAM_APIKEY>입니다. 여기서, <user_id>는 데이터 버킷에 액세스하는 데 apikey를 사용하는 사용자의 IBM Cloud ID입니다. API 키를 생성하려면 watsonx.data 콘솔에 로그인하고 프로필 > 프로필 및 설정 > API 키로 이동하여 새 API 키를 생성하세요.
• <OBJECT_NAME>: 이름 IBM Cloud Object Storage.
• <BUCKET_NAME>: 애플리케이션 파일이 있는 스토리지 버킷입니다.
• <COS_SERVICE_NAME>: 클라우드 개체 저장소 서비스 이름입니다.
• <python file name>: Spark 애플리케이션 파일 이름.
• <api_version> v2 API를 사용하는 경우 <api_version> 파라미터를 v2 으로 설정하고, v3 API의 경우 v3 으로 설정합니다.

제한사항:

• 사용자는 스키마 및 테이블을 만들 수 있는 전체 액세스 권한이 있어야 합니다.
• 데이터 정책을 만들려면 카탈로그를 Presto 엔진에 연결해야 합니다.
• 존재하지 않는 스키마를 표시하려고 하면 시스템에서 널 포인터 문제가 발생합니다.
• Iceberg, Hive Hudi 및 Delta Lake 카탈로그에 대해 Spark 액세스 제어 확장 기능을 활성화할 수 있습니다.