Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Connexions DynamoDB

Vous pouvez utiliser AWS Glue pour Spark pour lire et écrire dans des tables DynamoDB dans AWS Glue. Vous vous connectez à DynamoDB à l’aide des autorisations IAM associées à votre tâche AWS Glue. AWS Glue prend en charge l’écriture de données dans la table DynamoDB du compte AWS. Pour de plus amples informations, veuillez consulter Accès entre régions et entre comptes aux tables DynamoDB.

En plus du connecteur ETL DynamoDB AWS Glue, vous pouvez lire à partir de DynamoDB à l’aide du connecteur d’exportation DynamoDB, qui invoque une demande ExportTableToPointInTime DynamoDB et la stocke dans un emplacement Amazon S3 que vous fournissez, au format JSON DynamoDB. AWS Glue crée ensuite un objet DynamicFrame en lisant les données à partir de l’emplacement d’exportation Amazon S3.

Le dispositif d’écriture DynamoDB est disponible dans la version 1.0 ou les versions ultérieures de AWS Glue. Le dispositif de connecteur d’exportation DynamoDB AWS Glue est disponible dans la version 2.0 ou les versions ultérieures d’AWS Glue.

Pour plus d’informations sur DynamoDB, consultez la documentation Amazon DynamoDB.

Configuration de connexions DynamoDB

Pour vous connecter à DynamoDB à partir d’AWS Glue, accordez au rôle IAM associé à votre tâche AWS Glue l’autorisation d’interagir avec DynamoDB. Pour plus d’informations sur les autorisations requises pour lire ou écrire à partir de DynamoDB, consultez Actions, ressources et clés de condition pour DynamoDB dans la documentation IAM.

Dans les situations suivantes, vous autre peut-être besoin d’une configuration supplémentaire :

Lecture et écriture dans DynamoDB

Les exemples de code suivants montrent comment lire et écrire des tables DynamoDB. Ils montrent la lecture d'une table et l'écriture dans une autre table.

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

}

Utilisation du connecteur d’exportation DynamoDB

Le connecteur d'exportation fonctionne mieux que le connecteur ETL lorsque la taille de la table DynamoDB est supérieure à 80 Go. En outre, étant donné que la demande d'exportation est effectuée en dehors des processus Spark dans une tâche AWS Glue, vous pouvez activer l'auto scaling des tâches AWS Glue pour enregistrer l'utilisation du DPU pendant la demande d'exportation. Avec le connecteur d'exportation, vous n'avez pas non plus besoin de configurer le nombre de fractionnements pour le parallélisme de l'exécuteur Spark ou le pourcentage de lecture du débit DynamoDB.

Les exemples de code suivants montrent comment lire (via le connecteur d'exportation) et comment imprimer le nombre de partitions.

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

}

Ces exemples montrent comment lire (via le connecteur d'exportation) et comment imprimer le nombre de partitions à partir d'une table de catalogue de données AWS Glue qui comporte une classification 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())
)

Simplification de l’utilisation de l’exportation DynamoDB au format JSON

Les exportations DynamoDB avec le connecteur d’exportation AWS Glue DynamoDB donnent lieu à des fichiers JSON de structures imbriquées spécifiques. Pour plus d'informations, consultez Objets de données. AWS Gluefournit une transformation DynamicFrame, qui peut supprimer l'imbrication de telles structures dans un formulaire plus facile à utiliser pour les applications en aval.

La transformation peut être appelée de deux manières. Vous pouvez définir l’option de connexion "dynamodb.simplifyDDBJson" avec la valeur "true" lorsque vous appelez une méthode pour lire à partir de DynamoDB. Vous pouvez également appeler la transformation en tant que méthode disponible de façon indépendante dans la bibliothèque AWS Glue.

Tenez compte du schéma suivant généré par une exportation 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 transformation simplifyDDBJson simplifie cette exportation ainsi :

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

Configuration du parallélisme des opérations de DynamoDB

Pour améliorer les performances, vous pouvez régler certains paramètres disponibles pour le connecteur DynamoDB. Lorsque vous réglez les paramètres de parallélisme, votre objectif est d’optimiser l’utilisation des employés alloués dans AWS Glue. Ensuite, si vous avez besoin de plus de performances, nous vous recommandons de monter en puissance votre tâche en augmentant le nombre de DPU.

Vous pouvez modifier le parallélisme dans une opération de lecture de DynamoDB à l’aide du paramètre dynamodb.splits lorsque vous utilisez le connecteur ETL. Lors d’une tâche de lecture avec le connecteur d’exportation, vous n’avez pas besoin de configurer le nombre de fractionnements pour le parallélisme de l’exécuteur Spark. Vous pouvez modifier le parallélisme dans une opération d’écriture de DynamoDB avec dynamodb.output.numParallelTasks.

Lecture avec le connecteur ETL DynamoDB

Nous vous recommandons de calculer dynamodb.splits en fonction du nombre maximum d’employés défini dans la configuration de votre tâche et du calcul numSlots suivant. En cas de mise à l’échelle automatique, le nombre réel d’employés disponibles peut changer et passer en dessous de ce plafond. Pour plus d’informations sur la définition du nombre maximum d’employés, consultez Nombre d’employés (NumberOfWorkers) dans Configuration des propriétés des tâches pour les tâches Spark dans AWS Glue.

Nous vous recommandons de définir dynamodb.splits sur le nombre d’emplacements disponibles, numSlots.

Écriture dans DynamoDB

Le paramètre dynamodb.output.numParallelTasks est utilisé pour déterminer le WCU par tâche Spark, à l’aide du calcul suivant :

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

Le dispositif d’écriture DynamoDB fonctionne de manière optimale si la configuration représente avec précision le nombre de tâches Spark écrites dans DynamoDB. Dans certains cas, il se peut que vous ayez besoin de remplacer le calcul par défaut afin d’améliorer les performances d’écriture. Si vous ne spécifiez pas ce paramètre, le WCU autorisé pour chaque tâche Spark est automatiquement calculé par la formule suivante :

Référence des options de connexion DynamoDB

Désigne une connexion à Amazon DynamoDB.

Les options de connexion sont différentes pour une connexion source et une connexion collecteur.

« connectionType » : « dynamodb » avec le connecteur ETL comme source

Utilisez les options de connexion suivantes avec "connectionType": "dynamodb" en tant que source, lorsque vous utilisez le connecteur ETL DynamoDB AWS Glue :

« connectionType » : « dynamodb » avec le connecteur d'exportation DynamoDB AWS Glue comme source

Utilisez les options de connexion suivantes avec « connectionType » : « dynamodb » en tant que source, lorsque vous utilisez le connecteur d'exportation DynamoDB AWS Glue, disponible uniquement pour AWS Glue version 2.0 et au-delà :

« connectionType » : « dynamodb » avec le connecteur ETL comme récepteur

Utilisez les options de connexion suivantes avec "connectionType": "dynamodb" comme collecteur :