EMR サーバーレスOSSでの Delta Lake の使用 - Amazon EMR
Amazon EMRバージョン 6.9.0 以降Amazon EMRバージョン 6.8.0 以前

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

EMR サーバーレスOSSでの Delta Lake の使用

Amazon EMRバージョン 6.9.0 以降

注記

Amazon EMR 7.0.0 以降では、Delta Lake 3.0.0 が使用され、delta-core.jarファイル名が に変更されますdelta-spark.jar。Amazon 7.0.0 EMR 以降を使用する場合は、設定delta-spark.jarで を指定する必要があります。

Amazon EMR 6.9.0 以降には Delta Lake が含まれているため、Delta Lake を自分でパッケージ化したり、EMRサーバーレスジョブで--packagesフラグを指定したりする必要がなくなりました。

  1. EMR Serverless ジョブを送信するときは、次の設定プロパティがあることを確認し、次のパラメータを 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 テーブルの作成と読み取りをテストdelta_sample.pyするローカルを作成します。

    # 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 ファイルから Python ライブラリSparkContext.addPyFileを追加できます。

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 EMRバージョン 6.8.0 以前

Amazon EMR6.8.0 以前を使用している場合は、以下の手順に従ってEMR、サーバーレスアプリケーションで Delta Lake OSSを使用します。

  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