OSS 搭配 EMR Serverless 使用 Delta Lake - Amazon EMR
Amazon 6.9.0 版及更新EMR版本Amazon 6.8.0 版及更低EMR版本

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

OSS 搭配 EMR Serverless 使用 Delta Lake

Amazon 6.9.0 版及更新EMR版本

注意

Amazon EMR 7.0.0 及更高版本使用 Delta Lake 3.0.0，將delta-core.jar檔案重新命名為 delta-spark.jar。如果您使用 Amazon EMR 7.0.0 或更新版本，請務必在組態delta-spark.jar中指定 。

Amazon EMR 6.9.0 及更高版本包含 Delta Lake，因此您不再需要自行包裝 Delta Lake，或將--packages旗標與無EMR伺服器任務一起提供。

  1. 當您提交無EMR伺服器任務時，請確定您具有下列組態屬性，並在 sparkSubmitParameters 欄位中包含下列參數。

    --conf spark.jars=/usr/share/aws/delta/lib/delta-core.jar,/usr/share/aws/delta/lib/delta-storage.jar
    --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension
    --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog

  2. 建立本機delta_sample.py，以測試建立和讀取 Delta 資料表。

    # delta_sample.py
    from pyspark.sql import SparkSession
    
    import uuid
    
    url = "s3://amzn-s3-demo-bucket/delta-lake/output/%s/" % str(uuid.uuid4())
    spark = SparkSession.builder.appName("DeltaSample").getOrCreate()
    
    ## creates a Delta table and outputs to target S3 bucket
    spark.range(5).write.format("delta").save(url)
    
    ## reads a Delta table and outputs to target S3 bucket
    spark.read.format("delta").load(url).show

  3. 使用 AWS CLI，將delta_sample.py檔案上傳至您的 Amazon S3 儲存貯體。然後使用 start-job-run命令將任務提交至現有的 EMR Serverless 應用程式。

    aws s3 cp delta_sample.py s3://amzn-s3-demo-bucket/code/
    
    aws emr-serverless start-job-run \
        --application-id application-id \
        --execution-role-arn job-role-arn \
        --name emr-delta \
        --job-driver '{
            "sparkSubmit": {
                "entryPoint": "s3://amzn-s3-demo-bucket/code/delta_sample.py",
                "sparkSubmitParameters": "--conf spark.jars=/usr/share/aws/delta/lib/delta-core.jar,/usr/share/aws/delta/lib/delta-storage.jar --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"
            }
        }'

若要搭配 Delta Lake 使用 Python 程式庫，您可以透過封裝程式庫做為相依性，或使用程式庫做為自訂映像 來新增delta-core程式庫。

或者，您可以使用 從 delta-core JAR 檔案SparkContext.addPyFile新增 Python 程式庫：

import glob
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
spark.sparkContext.addPyFile(glob.glob("/usr/share/aws/delta/lib/delta-core_*.jar")[0])

Amazon 6.8.0 版及更低EMR版本

如果您使用的是 Amazon EMR 6.8.0 或更舊版本，請依照下列步驟將 Delta Lake OSS 與無EMR伺服器應用程式搭配使用。

  1. 若要在 Amazon EMR Serverless 應用程式中建置與 Spark 版本相容的 Delta Lake 開放原始碼版本，請導覽至 Delta GitHub 並遵循指示。

  2. 將 Delta Lake 程式庫上傳至您 中的 Amazon S3 儲存貯體 AWS 帳戶。

  3. 當您在應用程式組態中提交無EMR伺服器任務時，請包含目前位於儲存貯體中的 Delta Lake JAR 檔案。

    --conf spark.jars=s3://amzn-s3-demo-bucket/jars/delta-core_2.12-1.1.0.jar

  4. 若要確保您可以從 Delta 資料表讀取和寫入，請執行範例 PySpark測試。

    from pyspark import SparkConf, SparkContext
    from pyspark.sql import HiveContext, SparkSession
    
    import uuid
    
    conf = SparkConf()
    sc = SparkContext(conf=conf)
    sqlContext = HiveContext(sc)
    
    url = "s3://amzn-s3-demo-bucket/delta-lake/output/1.0.1/%s/" % str(uuid.uuid4())
    
    ## creates a Delta table and outputs to target S3 bucket
    session.range(5).write.format("delta").save(url)
    
    ## reads a Delta table and outputs to target S3 bucket
    session.read.format("delta").load(url).show