Conexões do DynamoDB

É possível usar o AWS Glue para Spark para ler e gravar em tabelas no DynamoDB no AWS Glue. Conecte-se ao DynamoDB usando as permissões do IAM anexadas ao seu trabalho do AWS Glue. O AWS Glueoferece suporte à gravação de dados em uma tabela do DynamoDB em uma conta da AWS diferente. Para ter mais informações, consulte Acesso a tabelas do DynamoDB entre contas e entre regiões.

Além do conector de ETL para AWS Glue DynamoDB, é possível ler do DynamoDB usando o conector de exportação para DynamoDB. Esse conector invoca uma solicitação ExportTableToPointInTime do DynamoDB e a armazena no formato DynamoDB JSON em um local do Amazon S3 fornecido por você. Em seguida, o AWS Glue cria um objeto DynamicFrame ao ler os dados do local de exportação do Amazon S3.

O gravador do DynamoDB é compatível com o AWS Glue versão 1.0 ou posterior. O conector de exportação do AWS Glue DynamoDB é compatível com o AWS Glue versão 2.0 ou posterior.

Para obter mais informações sobre o DynamoDB, consulte a documentação do Amazon DynamoDB.

Configurar conexões do DynamoDB

Para se conectar ao DynamoDB via AWS Glue, conceda ao perfil do IAM associado ao seu trabalho do AWS Glue permissão para interagir com o DynamoDB. Para obter mais informações sobre as permissões necessárias para ler ou gravar no DynamoDB, consulte Ações, recursos e chaves de condição para DynamoDB na documentação do IAM.

Nas seguintes situações, configurações adicionais podem ser necessárias:

Ler e gravar no DynamoDB

Os exemplos de código a seguir mostram como fazer a leitura (via conector ETL) e gravação de tabelas do DynamoDB. Eles demonstram a leitura de uma tabela e a gravação em uma outra tabela.

Python

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

dyf = glue_context.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={"dynamodb.input.tableName": test_source,
        "dynamodb.throughput.read.percent": "1.0",
        "dynamodb.splits": "100"
    }
)
print(dyf.getNumPartitions())

glue_context.write_dynamic_frame_from_options(
    frame=dyf,
    connection_type="dynamodb",
    connection_options={"dynamodb.output.tableName": test_sink,
        "dynamodb.throughput.write.percent": "1.0"
    }
)

job.commit()

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.input.tableName" -> test_source,
        "dynamodb.throughput.read.percent" -> "1.0",
        "dynamodb.splits" -> "100"
      ))
    ).getDynamicFrame()
    
    print(dynamicFrame.getNumPartitions())

    val dynamoDbSink: DynamoDbDataSink =  glueContext.getSinkWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.output.tableName" -> test_sink,
        "dynamodb.throughput.write.percent" -> "1.0"
      ))
    ).asInstanceOf[DynamoDbDataSink]
    
    dynamoDbSink.writeDynamicFrame(dynamicFrame)

    Job.commit()
  }

}

Usar o conector de exportação do DynamoDB

O conector de exportação apresenta um melhor desempenho do que o conector de ETL quando o tamanho da tabela do DynamoDB é superior a 80 GB. Além disso, como a solicitação de exportação é conduzida fora dos processos do Spark em um trabalho do AWS Glue, você pode habilitar a autoescalabilidade de trabalhos do AWS Glue para poupar o uso de DPU durante a solicitação de exportação. Com o conector de exportação, você também não precisa configurar o número de divisões para o paralelismo do executor do Spark ou o percentual de leitura de throughput do DynamoDB.

Os exemplos de código a seguir mostram como fazer a leitura (usando o conector de exportação) e imprimir o número de partições.

Python

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

dyf = glue_context.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={
        "dynamodb.export": "ddb",
        "dynamodb.tableArn": test_source,
        "dynamodb.s3.bucket": bucket_name,
        "dynamodb.s3.prefix": bucket_prefix,
        "dynamodb.s3.bucketOwner": account_id_of_bucket,
    }
)
print(dyf.getNumPartitions())

job.commit()

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.export" -> "ddb",
        "dynamodb.tableArn" -> test_source,
        "dynamodb.s3.bucket" -> bucket_name,
        "dynamodb.s3.prefix" -> bucket_prefix,
        "dynamodb.s3.bucketOwner" -> account_id_of_bucket,
      ))
    ).getDynamicFrame()
    
    print(dynamicFrame.getNumPartitions())

    Job.commit()
  }

}

Estes exemplos mostram como fazer a leitura (usando o conector de exportação) e imprimir o número de partições de uma tabela do AWS Glue Data Catalog que tenha uma classificação dynamodb:

Python

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

dynamicFrame = glue_context.create_dynamic_frame.from_catalog(
        database=catalog_database,
        table_name=catalog_table_name,
        additional_options={
            "dynamodb.export": "ddb", 
            "dynamodb.s3.bucket": s3_bucket,
            "dynamodb.s3.prefix": s3_bucket_prefix
        }
    )
print(dynamicFrame.getNumPartitions())

job.commit()

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getCatalogSource(
        database = catalog_database,
        tableName = catalog_table_name,
        additionalOptions = JsonOptions(Map(
            "dynamodb.export" -> "ddb", 
            "dynamodb.s3.bucket" -> s3_bucket,
            "dynamodb.s3.prefix" -> s3_bucket_prefix
        ))
    ).getDynamicFrame()
    print(dynamicFrame.getNumPartitions())
)

Simplificar o uso do JSON de exportação do DynamoDB

As exportações do DynamoDB feitas com o conector de exportação para DynamoDB do AWS Glue resultam em arquivos JSON de estruturas aninhadas específicas. Para mais informações, consulte Objetos de dados. O AWS Glue fornece uma transformação DynamicFrame, que pode desaninhar essas estruturas em uma forma mais fácil de usar para aplicações downstream.

A transformação pode ser invocada de duas formas. É possível definir a opção de conexão "dynamodb.simplifyDDBJson" com o valor "true" ao chamar um método para ler do DynamoDB. Você também pode chamar a transformação como um método disponível de forma independente na biblioteca do AWS Glue.

Considere o seguinte esquema gerado por uma exportação do DynamoDB:

root
|-- Item: struct
|    |-- parentMap: struct
|    |    |-- M: struct
|    |    |    |-- childMap: struct
|    |    |    |    |-- M: struct
|    |    |    |    |    |-- appName: struct
|    |    |    |    |    |    |-- S: string
|    |    |    |    |    |-- packageName: struct
|    |    |    |    |    |    |-- S: string
|    |    |    |    |    |-- updatedAt: struct
|    |    |    |    |    |    |-- N: string
|    |-- strings: struct
|    |    |-- SS: array
|    |    |    |-- element: string
|    |-- numbers: struct
|    |    |-- NS: array
|    |    |    |-- element: string
|    |-- binaries: struct
|    |    |-- BS: array
|    |    |    |-- element: string
|    |-- isDDBJson: struct
|    |    |-- BOOL: boolean
|    |-- nullValue: struct
|    |    |-- NULL: boolean

A transformação simplifyDDBJson simplificará isso para:

root
|-- parentMap: struct
|    |-- childMap: struct
|    |    |-- appName: string
|    |    |-- packageName: string
|    |    |-- updatedAt: string
|-- strings: array
|    |-- element: string
|-- numbers: array
|    |-- element: string
|-- binaries: array
|    |-- element: string
|-- isDDBJson: boolean
|-- nullValue: null

Configurar o paralelismo nas operações do DynamoDB

Para aumentar a performance, é possível ajustar alguns parâmetros disponíveis para o conector do DynamoDB. Seu objetivo ao ajustar os parâmetros de paralelismo é maximizar o uso dos operadores do AWS Glue provisionados. Assim, se você precisar de mais performance, recomendamos aumentar a escala do seu trabalho horizontalmente aumentando o número de DPUs.

Você pode alterar o paralelismo em uma operação de leitura do DynamoDB usando o parâmetro dynamodb.splits ao utilizar o conector ETL. Ao ler com o conector de exportação, não é necessário configurar o número de divisões para o paralelismo do executor do Spark. Você pode alterar o paralelismo em uma operação de gravação do DynamoDB com dynamodb.output.numParallelTasks.

Ler com o conector ETL do DynamoDB

Recomendamos calcular dynamodb.splits com base no número máximo de trabalhadores definido em sua configuração de trabalho e no cálculo numSlots a seguir. Em caso de escalonamento automático, o número real de operadores disponíveis pode cair abaixo desse limite. Para obter mais informações sobre como definir o número máximo de trabalhadores, consulte Número de trabalhadores (NumberOfWorkers) em Configurar propriedades de trabalho para trabalhos do Spark no AWS Glue.

Recomendamos definir dynamodb.splits como o número de slots disponíveis, numSlots.

Gravar no DynamoDB

O parâmetro dynamodb.output.numParallelTasks é usado para determinar a WCU por tarefa do Spark, usando o seguinte cálculo:

permittedWcuPerTask = ( TableWCU * dynamodb.throughput.write.percent ) / dynamodb.output.numParallelTasks

O gravador do DynamoDB funcionará melhor se a configuração representar com precisão o número de tarefas do Spark gravadas no DynamoDB. Em alguns casos, talvez seja necessário substituir o cálculo padrão para aumentar o desempenho de gravação. Se você não especificar esse parâmetro, as WCU permitidas por tarefa do Spark serão calculadas automaticamente pela seguinte fórmula:

Referência de opções de conexão do DynamoDB

Designa uma conexão com o Amazon DynamoDB.

As opções de conexão diferem para uma conexão da fonte e uma conexão do coletor.

“connectionType”: “dynamodb” com o conector de ETL como fonte

Ao usar o conector de ETL para DynamoDB do AWS Glue, use as seguintes opções de conexão com "connectionType": "dynamodb" como fonte:

“connectionType”: “dynamodb” com o conector de exportação para DynamoDB do AWS Glue como fonte

Use as seguintes opções de conexão com "connectionType": "dynamodb" como fonte ao usar o conector de exportação para DynamoDB do AWS Glue, que está disponível apenas para o AWS Glue versão 2.0 em diante:

“connectionType”: “dynamodb” com o conector de ETL como coletor

Use as seguintes opções de conexão com "connectionType": "dynamodb" como coletor: