Conexiones de DynamoDB

Puede utilizar AWS Glue para Spark para leer y escribir en tablas de DynamoDB en AWS Glue. Para conectarse a DynamoDB, utilice los permisos de IAM adjuntos a su trabajo de AWSGlue. AWS Glue admite la escritura de datos en otra cuenta de tablas de DynamoDB de AWS. Para obtener más información, consulte Acceso entre cuentas y entre regiones a tablas de DynamoDB.

Además del conector de ETL de DynamoDB de AWS Glue, puede leer en DynamoDB por medio del conector de exportación de DynamoDB, que invoca una solicitud ExportTableToPointInTime de DynamoDB y la almacena en una ubicación de Amazon S3 que se proporcione, con el formato de DynamoDB JSON. Después, AWS Glue crea un objeto DynamicFrame mediante la lectura de los datos desde la ubicación de exportación de Amazon S3.

El escritor de DynamoDB está disponible en la versión AWS Glue 1.0 o versiones posteriores. El conector de exportación de DynamoDB de AWS Glue está disponible en la versión 2.0 o versiones posteriores de AWS Glue.

Para obtener más información sobre DynamoDB, consulte la documentación de Amazon DynamoDB.

Configuración de las conexiones DynamoDB

Para conectarse a DynamoDB desde AWS Glue, conceda permiso al rol de IAM asociado a su trabajo de AWS Glue para interactuar con DynamoDB. Para obtener más información sobre los permisos necesarios para leer o escribir en DynamoDB, consulte Acciones, recursos y claves de condición para DynamoDB en la documentación de IAM.

En las siguientes situaciones, es posible que necesite una configuración adicional:

Lectura y escritura en DynamoDB

En los siguientes ejemplos de código, se muestra cómo leer (con el conector de ETL) de tablas de DynamoDB y escribir en ellas. Demuestran la lectura de una tabla y la escritura en otra tabla.

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()
  }

}

Uso del conector de exportación de DynamoDB

El conector de exportación funciona mejor que el conector de ETL cuando el tamaño de la tabla de DynamoDB es superior a 80 GB. Además, dado que la solicitud de exportación se lleva a cabo fuera de los procesos de Spark en un trabajo de AWS Glue, se puede habilitar el escalado automático de los trabajos de AWS Glue para guardar el uso de DPU durante la solicitud de exportación. Con el conector de exportación, tampoco es necesario configurar el número de divisiones del paralelismo del ejecutor de Spark o el porcentaje de lectura de rendimiento de DynamoDB.

Los siguientes ejemplos de código muestran cómo leer desde particiones (a través del conector de exportación) e imprimir el número de estas.

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()
  }

}

Estos ejemplos muestran cómo realizar la lectura desde particiones (a través del conector de exportación) e imprimir el número de estas desde una tabla de Data Catalog de AWS Glue que tenga una clasificación 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())
)

Simplificación del uso del JSON de exportación de DynamoDB

Las exportaciones de DynamoDB con el conector de exportación de DynamoDB de AWS Glue generan archivos JSON con estructuras anidadas específicas. Para obtener más información, consulte Objetos de datos. AWS Glue proporciona una transformación de DynamicFrame, que puede desanidar estas estructuras y dejarlas con una forma más fácil de usar para aplicaciones posteriores.

La transformación se puede invocar de dos formas. Puede configurar la opción de conexión de "dynamodb.simplifyDDBJson" con el valor "true" cuando se llame a un método para leer desde DynamoDB. También puede llamar a la transformación como un método disponible de forma independiente en la biblioteca de AWS Glue.

Considere el siguiente esquema generado por una exportación a 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

La transformación de simplifyDDBJson simplificará esto de la siguiente forma:

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

Configuración del paralelismo en las operaciones de DynamoDB

Para mejorar el rendimiento, puede ajustar determinados parámetros disponibles para el conector DynamoDB. Su objetivo al ajustar los parámetros de paralelismo es maximizar el uso de los trabajadores de AWS Glue aprovisionados. Por lo tanto, si necesita más rendimiento, le recomendamos que amplíe su trabajo al aumentar el número de DPU.

Puede modificar el paralelismo en una operación de lectura de DynamoDB mediante el parámetro dynamodb.splits cuando utilice el conector ETL. Cuando lea con el conector de exportación, no es necesario configurar el número de divisiones del paralelismo del ejecutor de Spark. Puede modificar el paralelismo en una operación de escritura de DynamoDB con dynamodb.output.numParallelTasks.

Lectura con el conector ETL de DynamoDB

Le recomendamos que calcule dynamodb.splits en función del número máximo de trabajadores establecido en la configuración de su trabajo y del siguiente cálculo numSlots. Si se autoescala, la cantidad real de trabajadores disponibles puede cambiar por debajo de ese límite. Para obtener más información sobre cómo establecer el número máximo de trabajadores, consulte Número de trabajadores (NumberOfWorkers) en Configuración de las propiedades de trabajos para trabajos de Spark en AWS Glue.

Le recomendamos que establezca dynamodb.splits al número de ranuras disponibles, numSlots.

Escritura en DynamoDB

El parámetro dynamodb.output.numParallelTasks se utiliza para determinar la WCU por tarea de Spark mediante el siguiente cálculo:

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

El escritor de DynamoDB funcionará mejor si la configuración representa con precisión el número de tareas de Spark que se escriben en DynamoDB. En algunos casos, es posible que necesite anular el cálculo predeterminado para mejorar el rendimiento de la escritura. Si no especifica este parámetro, el WCU permitido por tarea de Spark se calculará automáticamente mediante la siguiente fórmula:

Referencia de opciones de conexión de DynamoDB

Designa una conexión a Amazon DynamoDB.

Las opciones de conexión son distintas entre una conexión de origen y una conexión de receptor.

"connectionType": "dynamodb" con el conector de ETL como origen

Utilice las siguientes opciones de conexión con "connectionType": "dynamodb" como origen cuando use el conector de ETL de DynamoDB de AWS Glue:

“connectionType”: “dynamodb” con el conector de exportación de DynamoDB de AWS Glue como origen

Utilice las siguientes opciones de conexión con “connectionType”: “dynamodb” como origen cuando use el conector de exportación de DynamoDB de AWS Glue, que solo está disponible a partir de AWS Glue versión 2.0:

“connectionType”: “dynamodb” con el conector de ETL como receptor

Utilice las siguientes opciones de conexión con "connectionType": "dynamodb" como receptor: