使用 Spark 存取控制延伸增強 Spark 應用程式的提交

適用於：火花機

當您提交使用註冊的外部儲存桶的 Spark 應用程式時watsonx.data，Spark存取控制擴充允許額外的授權從而增強安全性。如果在spark配置中啟用該擴展，則只有授權使用者才可以存取和操作watsonx.data透過 Spark 作業進行目錄。

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

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

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

若要提交 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>：存取上述資料桶的端點的主機名稱。例子，s3.us-south。cloud-object-storage.appdomain.cloud用於美國南部區域的雲端對象儲存桶。
<COS_SERVICE_NAME>:提供 Cloud object Storage 服務名稱。
<COS_ENDPOINT> 提供公共端點。如需詳細資訊，請參閱 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 存取控制延伸功能。