EMR ServerlessOSS에서 Delta Lake 사용 - Amazon EMR
Amazon EMR 버전 6.9.0 이상Amazon EMR 버전 6.8.0 이하

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

EMR ServerlessOSS에서 Delta Lake 사용

Amazon EMR 버전 6.9.0 이상

참고

Amazon EMR 7.0.0 이상은 delta-core.jar 파일 이름을 로 바꾸는 Delta Lake 3.0.0을 사용합니다delta-spark.jar. Amazon EMR 7.0.0 이상을 사용하는 경우 구성delta-spark.jar에서 를 지정해야 합니다.

Amazon EMR 6.9.0 이상에는 Delta Lake가 포함되어 있으므로 더 이상 Delta Lake를 직접 패키징하거나 EMR Serverless 작업과 함께 --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_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. 를 사용하여 delta_sample.py 파일을 Amazon S3 버킷에 AWS CLI업로드합니다. 그런 다음 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"
            }
        }'

Python 라이브러리를 Delta Lake와 함께 사용하려면 라이브러리를 종속성으로 패키징하거나 사용자 지정 이미지로 사용하여 delta-core 라이브러리를 추가할 수 있습니다. https://docs.aws.amazon.com/emr/latest/EMR-Serverless-UserGuide/using-custom-images.html

또는 SparkContext.addPyFile를 사용하여 delta-core JAR 파일에서 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 EMR 버전 6.8.0 이하

Amazon EMR 6.8.0 이하를 사용하는 경우 다음 단계에 따라 OSS EMR 서버리스 애플리케이션에서 Delta Lake를 사용합니다.

  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