使用 Spark 存取控制擴充增強 Spark 應用程式提交

當您提交使用在中註冊的外部儲存桶的 Spark 應用程式時，watsonx.data Spark 存取控制擴充功能可提供額外的授權機制，從而強化安全性。若在 Spark 設定中啟用此擴充功能，僅授權使用者可透過 Spark 工作存取及操作 watsonx.data 目錄。

在中註冊外部 Spark watsonx.data 引擎的選項已遭棄用，並將於版本中移除。watsonx.data 現已內建多種 Spark 引擎供您直接配置使用，包含 2.3 Gluten 加速版 Spark 引擎與原生 watsonx.data Spark 引擎。

您可以為 Iceberg、Hudi Hive 和 Delta Lake 目錄啟用 Spark 存取控制擴充功能。

您可以使用 Ranger 或存取管理系統 (AMS) 資料政策來授予或拒絕使用者、使用者群組、目錄 (Iceberg, Hive, Hudi 和 Delta Lake )、模式、資料表和欄位的存取權限。除了資料級授權外，還考慮儲存權限。有關在目錄(Iceberg, Hive, Hudi 和 Delta Lake )、資料桶、模式和資料表使用 AMS 的詳細資訊，請參閱管理角色和權限。有關如何建立 Ranger 策略（在 Hadoop SQL 服務下定義）以及在目錄(Iceberg, Hive, Hudi and Delta Lake )、資料桶、模式和資料表上啟用這些策略的詳細資訊，請參閱管理 Ranger 策略。

必要條件

創造Cloud Object Storage儲存 Spark 應用程式中使用的資料。創造Cloud Object Storage和一個桶，看建立儲存桶。您可以配置兩個儲存桶，資料儲存桶來存儲watsonx.data用於維護 Spark 應用程式程式碼的表和應用程式儲存桶。
登記Cloud Object Storage桶裡watsonx.data。有關更多信息，請參閱新增儲存桶目錄對。
將Spark應用程式上傳到存儲，參見上傳數據。
您必須具有 IAM 管理員角色或MetastoreAdmin角色，用於在內部建立模式或表watsonx.data。

程序

Spark存取控制擴充支援外部Spark引擎。

若要啟用 Spark 存取控制擴展，您必須使用下列命令更新 Spark 配置 add authz.IBMSparkACExtension to spark.sql.extensions。
儲存以下內容Python申請為iceberg.py。

冰山被認為是一個例子。您亦可使用 Hudi Hive 和 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()

若要提交 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>",
  }
}

自 watsonx.data 版本起，使用 ibmlhapikey 和 ibmlhtoken 作為使用者 2.2.0 名稱的驗證方式已廢棄。這些格式將在 2.3.0 本次版本中逐步淘汰。為確保與即將推出的版本相容，請使用新格式：ibmlhapikey_<username> 和 ibmlhtoken_<username>。

參數值：

<region>：配置實例的區域。例如，us-south 區域。
<spark_engine_id> Spark 實例的唯一識別碼。有關如何擷取 ID 的資訊，請參閱管理 Spark 引擎詳細資訊。
<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>：從基礎架構管理員關聯到 spark 引擎的資料桶名稱。
<wxd-data-bucket-endpoint>：存取上述資料桶的端點的主機名稱。例子，s3.us-south。cloud-object-storage.appdomain.cloud用於美國南部區域的雲端對象儲存桶。
<wxd-bucket-catalog-name>：與資料桶關聯的目錄名稱。
<wxd-catalog-metastore-host>：與註冊的儲存桶關聯的元儲存。
<cos_bucket_endpoint>：提供 Metastore 主機值。有關更多信息，請參閱儲存詳細信息。
<access_key>：提供access_key_id。有關更多信息，請參閱儲存詳細信息。
<secret_key>：提供secret_access_key。有關更多信息，請參閱儲存詳細信息。
<truststore_path>：提供trustore憑證上傳的COS路徑。例如 cos://di-bucket.di-test/1902xx-truststore.jks。有關生成 trustore 的更多信息，請參閱導入自簽名憑證。
<cas_endpoint>：資料存取服務 (DAS) 端點。若要取得 DAS 端點，請參閱取得 DAS 端點。
<username>：您的用戶名watsonx.data實例。在這裡，ibmlhapikey。
<apikey>：這base64編碼`ibmlhapikey_<user_id>:<IAM_APIKEY>。這裡，<user_id> 是IBM Cloud使用apikey存取資料桶的使用者id。若要產生 API 金鑰，請登入watsonx.data控制台並導覽至設定檔 > 設定檔和設定 > API 金鑰並產生新的 API 金鑰。
<OBJECT_NAME> 名稱 IBM Cloud Object Storage。
<BUCKET_NAME>：應用程式檔案所在的儲存桶。
<COS_SERVICE_NAME>：雲端對象儲存服務名稱。
<python file name> 火花應用程式檔案名稱。
<api_version>:使用 v2 API 時，請將 <api_version> 參數設定為 v2；使用 v3 API 時，請將它設定為 v3。

限制：

使用者必須具有完全存取權限才能建立架構和表。
若要建立資料策略，您必須將目錄關聯到Presto引擎。
如果您嘗試顯示不存在的架構，系統會拋出空指標問題。
您可以為 Iceberg、Hudi Hive 和 Delta Lake 目錄啟用 Spark 存取控制擴充功能。