전제조건

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

프로시저

Spark 액세스 제어 확장 프로그램은 기본 Spark 엔진을 지원합니다.

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

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

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

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.hadoop.wxd.apiKey":"Basic xxx",
        "spark.sql.extensions":"<required-storage-support-extension>,authz.IBMSparkACExtension"

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

매개변수 값:

• <token>: 서비스 인스턴스에 대한 액세스 토큰을 가져옵니다. 토큰 생성에 대한 자세한 정보는 토큰 생성을 참고하세요.
• <instance_id> watsonx.data 클러스터 인스턴스 URL 인스턴스 ID입니다. 예: crn:v1:staging:public:lakehouse:us-south:a/7bb9e380dc0c4bc284592b97d5095d3c:5b602d6a-847a-469d-bece-0a29124588c0::.
• <wxd-data-bucket-endpoint>: 위에서 언급한 데이터 버킷에 액세스하기 위한 엔드포인트의 호스트 이름입니다. 예를 들어, us-south 리전 클라우드 오브젝트 스토리지 버킷의 경우 s3.us-south.cloud-object-storage.appdomain.cloud로 설정합니다.
• <COS_SERVICE_NAME>: 클라우드 객체 저장 서비스 이름을 입력하세요.
• <COS_ENDPOINT> 공개 엔드포인트를 입력합니다. 더 자세한 정보를 원하시면 엔드포인트를 참고하세요.
• <access_key>: access_key_id를 입력합니다. 자세한 정보는 자격 증명을 참고하세요.
• <secret_key>: secret_access_key를 입력합니다. 자세한 정보는 자격 증명을 참고하세요.
• <BUCKET_NAME>: 애플리케이션 파일이 있는 스토리지 버킷입니다.
• <python_file_name>: Spark 애플리케이션 파일 이름입니다.
• <api_version> v2 API를 사용하는 경우 <api_version> 파라미터를 v2 으로 설정하고, v3 API의 경우 v3 으로 설정합니다.

제한사항:

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