Connexions Amazon DocumentDB

Vous pouvez utiliser AWS Glue pour Spark pour lire et écrire dans les tables des bases de données Amazon DocumentDB. Vous pouvez vous connecter à Amazon DocumentDB à l’aide des informations d’identification enregistrées dans AWS Secrets Manager via une connexion AWS Glue.

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

Note

Les clusters élastiques Amazon DocumentDB ne sont actuellement pas pris en charge lors de l'utilisation du connecteur AWS Glue. Pour plus d'informations sur les clusters élastiques, consultez Using Amazon DocumentDB elastic clusters.

Lecture et écriture dans les collections Amazon DocumentDB

Note

Lorsque vous créez une tâche ETL qui se connecte à Amazon DocumentDB pour la propriété de tâche Connections, vous devez désigner un objet de connexion qui spécifie le virtual private cloud (VPC) dans lequel Amazon DocumentDB est exécuté. Pour l'objet de connexion, le type de connexion doit être JDBC et le JDBC URL doit être mongo://<DocumentDB_host>:27017.

Note

Ces exemples de code ont été développés pour AWS Glue 3.0. Pour migrer vers AWS Glue 4.0, consultez MongoDB. Le paramètre uri a été modifié.

Note

Lorsque vous utilisez Amazon DocumentDB, le paramètre retryWrites doit être défini sur false dans certaines situations, par exemple lorsque le document écrit spécifie _id. Pour plus d’informations, consultez Différences fonctionnelles avec MongoDB dans la documentation Amazon DocumentDB.

Le script Python suivant illustre l’utilisation des types et des options de connexion pour la lecture et l’écriture dans Amazon DocumentDB.

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext, SparkConf
from awsglue.context import GlueContext
from awsglue.job import Job
import time

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

job = Job(glueContext)
job.init(args['JOB_NAME'], args)

output_path = "s3://some_bucket/output/" + str(time.time()) + "/"
documentdb_uri = "mongodb://<mongo-instanced-ip-address>:27017"
documentdb_write_uri = "mongodb://<mongo-instanced-ip-address>:27017"

read_docdb_options = {
    "uri": documentdb_uri,
    "database": "test",
    "collection": "coll",
    "username": "username",
    "password": "1234567890",
    "ssl": "true",
    "ssl.domain_match": "false",
    "partitioner": "MongoSamplePartitioner",
    "partitionerOptions.partitionSizeMB": "10",
    "partitionerOptions.partitionKey": "_id"
}

write_documentdb_options = {
    "retryWrites": "false",
    "uri": documentdb_write_uri,
    "database": "test",
    "collection": "coll",
    "username": "username",
    "password": "pwd"
}

# Get DynamicFrame from  DocumentDB
dynamic_frame2 = glueContext.create_dynamic_frame.from_options(connection_type="documentdb",
                                                               connection_options=read_docdb_options)

# Write DynamicFrame to MongoDB and DocumentDB
glueContext.write_dynamic_frame.from_options(dynamic_frame2, connection_type="documentdb",
                                             connection_options=write_documentdb_options)

job.commit()

Le script Scala suivant illustre l’utilisation des types et des options de connexion pour la lecture et l’écriture dans Amazon DocumentDB.

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.MappingSpec
import com.amazonaws.services.glue.errors.CallSite
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.DynamicFrame
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._

object GlueApp {
  val DOC_URI: String = "mongodb://<mongo-instanced-ip-address>:27017"
  val DOC_WRITE_URI: String = "mongodb://<mongo-instanced-ip-address>:27017"
  lazy val documentDBJsonOption = jsonOptions(DOC_URI)
  lazy val writeDocumentDBJsonOption = jsonOptions(DOC_WRITE_URI)
  def main(sysArgs: Array[String]): Unit = {
    val spark: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)

    // Get DynamicFrame from DocumentDB
    val resultFrame2: DynamicFrame = glueContext.getSource("documentdb", documentDBJsonOption).getDynamicFrame()

    // Write DynamicFrame to DocumentDB
    glueContext.getSink("documentdb", writeJsonOption).writeDynamicFrame(resultFrame2)

    Job.commit()
  }

  private def jsonOptions(uri: String): JsonOptions = {
    new JsonOptions(
      s"""{"uri": "${uri}",
         |"database":"test",
         |"collection":"coll",
         |"username": "username",
         |"password": "pwd",
         |"ssl":"true",
         |"ssl.domain_match":"false",
         |"partitioner": "MongoSamplePartitioner",
         |"partitionerOptions.partitionSizeMB": "10",
         |"partitionerOptions.partitionKey": "_id"}""".stripMargin)
  }
}

Référence des options de connexion Amazon DocumentDB

Désigne une connexion à Amazon DocumentDB (compatible avec MongoDB).

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

« connectionType »: « documentdb » comme cource

Utilisez les options de connexion suivantes avec "connectionType": "documentdb" comme source :

• "uri" : (obligatoire) hôte Amazon DocumentDB à lire, dans le format mongodb://<host>:<port>.

• "database" : (obligatoire) base de données Amazon DocumentDB à lire.

• "collection" : (obligatoire) collection Amazon DocumentDB à lire.

• "username" : (obligatoire) nom d'utilisateur Amazon DocumentDB.

• "password" : (obligatoire) mot de passe Amazon DocumentDB.

• "ssl" : (obligatoire si vous utilisez SSL) si votre connexion utilise SSL, vous devez inclure cette option avec la valeur "true".

• "ssl.domain_match" : (obligatoire si vous utilisez SSL) si votre connexion utilise SSL, vous devez inclure cette option avec la valeur "false".

• "batchSize" : (Facultatif) : Nombre de documents à renvoyer par lot, utilisé dans le curseur des lots internes.

• "partitioner" : (facultatif) : nom de classe de l'outil du partitionneur pour lire les données d'entrée à partir d'Amazon DocumentDB. Le connecteur fournit les partitionneurs suivants :

  • MongoDefaultPartitioner (par défaut) (non pris en charge dans AWS Glue 4.0)

  • MongoSamplePartitioner (non pris en charge dans AWS Glue 4.0)

  • MongoShardedPartitioner

  • MongoSplitVectorPartitioner

  • MongoPaginateByCountPartitioner

  • MongoPaginateBySizePartitioner (non pris en charge dans AWS Glue 4.0)

• "partitionerOptions" ( Facultatif) : Options pour le partitionneur désigné. Les options suivantes sont prises en charge pour chaque partitionneur :

  • MongoSamplePartitioner: partitionKey, partitionSizeMB, samplesPerPartition

  • MongoShardedPartitioner: shardkey

  • MongoSplitVectorPartitioner : partitionKey, partitionSizeMB

  • MongoPaginateByCountPartitioner: partitionKey, numberOfPartitions

  • MongoPaginateBySizePartitioner : partitionKey, partitionSizeMB

  Pour plus d'informations sur ces options, consultez Configuration du partitionneur dans la documentation MongoDB.

« connectionType »: « documentdb » comme collecteur

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

• "uri" : (obligatoire) hôte Amazon DocumentDB où écrire, dans le format mongodb://<host>:<port>.

• "database" : (obligatoire) base de données Amazon DocumentDB où écrire.

• "collection" : (obligatoire) collection Amazon DocumentDB où écrire.

• "username" : (obligatoire) nom d'utilisateur Amazon DocumentDB.

• "password" : (obligatoire) mot de passe Amazon DocumentDB.

• "extendedBsonTypes" : (facultatif) si la valeur est true, cela autorise les types BSON étendus lors de l'écriture de données dans Amazon DocumentDB. La valeur par défaut est true.

• "replaceDocument" : (Facultatif) Si true, remplace l'ensemble du document lors de l'enregistrement de jeux de données contenant un champ _id. Si false, seuls les champs du document qui correspondent aux champs du jeu de données sont mis à jour. La valeur par défaut est true.

• "maxBatchSize" : (Facultatif) : Taille maximale du lot pour les opérations en bloc lors de l'enregistrement des données. La valeur par défaut est 512.

• "retryWrites" : (Facultatif) : Réessayer automatiquement certaines opérations d'écriture une seule fois si AWS Glue rencontre une erreur réseau.