Usar o Delta Lake OSS com o EMR Sem Servidor - Amazon EMR
Amazon EMR 6.9.0 e versões posterioresAmazon EMR 6.8.0 e versões anteriores

Usar o Delta Lake OSS com o EMR Sem Servidor

Amazon EMR 6.9.0 e versões posteriores

nota

O Amazon EMR 7.0.0 e versões superiores usam o Delta Lake 3.0.0, que renomeia o arquivo delta-core.jar como delta-spark.jar. Se você usa o Amazon EMR 7.0.0 ou superior, certifique-se de especificar delta-spark.jar nas configurações.

O Amazon EMR 6.9.0 e versões posteriores incluem o Delta Lake, então você não precisa mais empacotar o Delta Lake por conta própria ou fornecer o sinalizador --packages com trabalhos do EMR Sem Servidor.

  1. Ao enviar trabalhos do EMR Sem Servidor, certifique-se de ter as propriedades de configuração a seguir e incluir os parâmetros no campo 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. Crie um delta_sample.py local para testar a criação e a leitura de uma tabela do 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. Usando a AWS CLI, faça upload do arquivo delta_sample.py no bucket do Amazon S3. Em seguida, use o comando start-job-run para enviar um trabalho a uma aplicação existente do EMR Sem Servidor.

    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"
            }
        }'

Para usar bibliotecas do Python com o Delta Lake, você pode adicionar a biblioteca delta-core empacotando-a como uma dependência ou usando-a como uma imagem personalizada.

Como alternativa, você pode usar SparkContext.addPyFile para adicionar as bibliotecas do Python do arquivo JAR delta-core:

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 e versões anteriores

Se estiver usando o Amazon EMR 6.8.0 ou inferior, siga estas etapas para usar o OSS do Delta Lake com as aplicações do EMR Sem Servidor.

  1. Para criar uma versão de código aberto do Delta Lake que seja compatível com a versão do Spark na aplicação do Amazon EMR Sem Servidor, navegue até o Delta GitHub e siga as instruções.

  2. Faça upload das bibliotecas do Delta Lake em um bucket do Amazon S3 na Conta da AWS.

  3. Ao enviar trabalhos do EMR Sem Servidor na configuração da aplicação, inclua os arquivos JAR do Delta Lake que agora estão no bucket.

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

  4. Para garantir que você possa ler e gravar em uma tabela do Delta, execute um exemplo de teste do 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