Connessioni MongoDB

È possibile utilizzare AWS Glue per Spark per leggere e scrivere su tabelle in MongoDB w MongoDB Atlas in AWS Glue 4.0 e versioni successive. È possibile connettersi a MongoDB utilizzando le credenziali di nome utente e password memorizzate in AWS Secrets Manager tramite una connessione AWS Glue.

Per ulteriori informazioni su MongoDB, consulta la documentazione di MongoDB.

Configurazione delle connessioni MongoDB

Per connetterti a MongoDB da AWS Glue, avrai bisogno delle tue credenziali MongoDB, mongodbUser e mongodbPass.

Per connettersi a MongoDB da AWS Glue, potrebbero essere necessari alcuni prerequisiti:

• Se la tua istanza MongoDB si trova in un Amazon VPC, configura Amazon VPC per consentire al processo AWS Glue di comunicare con l'istanza MongoDB senza che il traffico attraversi la rete Internet pubblica.

  In Amazon VPC, identifica o crea un VPC, una sottorete e un gruppo di sicurezza che AWS Glue utilizzerà durante l'esecuzione del processo. Inoltre, assicurati che Amazon VPC sia configurato per consentire il traffico di rete tra l'istanza MongoDB e questa posizione. In base al layout di rete, potrebbe richiedere modifiche alle regole dei gruppi di sicurezza, alle liste di controllo accessi di rete, ai gateway NAT e alle connessioni peering.

È quindi possibile procedere alla configurazione di AWS Glue per l'uso con MongoDB.

Per configurare una connessione a MongoDB:

1. Facoltativamente, in AWS Secrets Manager, crea un segreto utilizzando le tue credenziali MongoDB. Per creare un segreto in Secrets Manager, segui il tutorial disponibile in Create an AWS Secrets Manager secret nella documentazione di AWS Secrets Manager. Dopo aver creato il segreto, prendi nota del nome, secretName, per il passaggio successivo.

   • Quando selezioni le coppie chiave/valore, crea una coppia per la chiave username con il valore mongodbUser.

     Quando selezioni le coppie chiave/valore, crea una coppia per la chiave password con il valore mongodbPass.

2. Nella console AWS Glue, crea una connessione seguendo i passaggi riportati in Aggiunta di una connessione AWS Glue. Dopo aver creato la connessione, prendi nota del nome, connectionName, per l'uso futuro in AWS Glue.

   • Quando selezioni un tipo di connessione, seleziona MongoDB o MongoDB Atlas.

   • Quando selezioni l'URL MongoDB o URL MongoDB Atlas, fornisci il nome host dell'istanza MongoDB.

     Un URL MongoDB viene fornito nel formato mongodb://mongoHost:mongoPort/mongoDBname.

     Un URL MongoDB Atlas viene fornito nel formato mongodb+srv://mongoHost:mongoPort/mongoDBname.

     Fornendo il database predefinito per la connessione, mongoDBname è facoltativo.

   • Se hai scelto di creare un segreto di Secrets Manager, scegli il tipo di credenziali AWS Secrets Manager.

     Quindi, in Segreto di AWS, fornisci secretName.

   • Se scegli di fornire nome utente e password, fornisci mongodbUser e mongodbPass.

3. Nelle seguenti situazioni, potresti aver bisogno di una configurazione aggiuntiva:

   • Per le istanze MongoDB ospitate su AWS in un Amazon VPC

     • Dovrai fornire le informazioni di connessione Amazon VPC alla connessione AWS Glue che definisce le credenziali di sicurezza MongoDB. Durante la creazione o l'aggiornamento della connessione, imposta VPC, sottorete e Gruppi di sicurezza nelle opzioni di rete.

Dopo aver creato una connessione MongoDB AWS Glue, dovrai eseguire i seguenti passaggi prima di chiamare il metodo di connessione.

• Se hai scelto di creare un segreto di Secrets Manager, concedi al ruolo IAM associato al processo AWS Glue il permesso di leggere secretName.

• Nella configurazione del processo AWS Glue, fornisci connectionName come Connessione di rete aggiuntiva.

Per utilizzare la connessione MongoDB AWS Glue in AWS Glue per Spark, fornisci l'opzione connectionName nella chiamata al metodo di connessione. In alternativa, puoi seguire i passaggi in Lavorare con le connessioni MongoDB nei processi ETL per utilizzare la connessione insieme al Catalogo dati AWS Glue.

Lettura da MongoDB utilizzando una connessione AWS Glue

Prerequisiti:

• Una raccolta MongoDB da cui desideri leggere. Avrai bisogno delle informazioni di identificazione per la raccolta.

  Una raccolta MongoDB è identificata da un nome di database e di raccolta, mongodbName, mongodbCollection.

• Una connessione MongoDB AWS Glue configurata per fornire informazioni di autenticazione. Completa i passaggi della procedura precedente, Per configurare una connessione a MongoDB per configurare le informazioni di autenticazione. Sarà necessario il nome della connessione AWS Glue, connectionName.

Ad esempio:

mongodb_read = glueContext.create_dynamic_frame.from_options(
    connection_type="mongodb",
    connection_options={
        "connectionName": "connectionName",
        "database": "mongodbName",
        "collection": "mongodbCollection",
        "partitioner": "com.mongodb.spark.sql.connector.read.partitioner.SinglePartitionPartitioner",
        "partitionerOptions.partitionSizeMB": "10",
        "partitionerOptions.partitionKey": "_id",
        "disableUpdateUri": "false",
    }
)

Scrittura su tabelle MongoDB

Questo esempio scrive informazioni a partire da un DynamicFrame esistente, dynamicFrame a MongoDB.

Prerequisiti:

• Una raccolta MongoDB su cui desideri scrivere. Avrai bisogno delle informazioni di identificazione per la raccolta.

  Una raccolta MongoDB è identificata da un nome di database e di raccolta, mongodbName, mongodbCollection.

• Una connessione MongoDB AWS Glue configurata per fornire informazioni di autenticazione. Completa i passaggi della procedura precedente, Per configurare una connessione a MongoDB per configurare le informazioni di autenticazione. Sarà necessario il nome della connessione AWS Glue, connectionName.

Ad esempio:

glueContext.write_dynamic_frame.from_options(
    frame=dynamicFrame,
    connection_type="mongodb",
    connection_options={
        "connectionName": "connectionName",
        "database": "mongodbName",
        "collection": "mongodbCollection",
        "disableUpdateUri": "false",
        "retryWrites": "false", 
    },
)

Lettura e scrittura in tabelle su tabelle MongoDB

Questo esempio scrive informazioni a partire da un DynamicFrame esistente, dynamicFrame a MongoDB.

Prerequisiti:

• Una raccolta MongoDB da cui desideri leggere. Avrai bisogno delle informazioni di identificazione per la raccolta.

  Una raccolta MongoDB su cui desideri scrivere. Avrai bisogno delle informazioni di identificazione per la raccolta.

  Una raccolta MongoDB è identificata da un nome di database e di raccolta, mongodbName, mongodbCollection.

• Informazioni di autenticazione MongoDB, mongodbUser e mongodbPassword.

Ad esempio:

Python

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()) + "/"
mongo_uri = "mongodb://<mongo-instanced-ip-address>:27017"
mongo_ssl_uri = "mongodb://<mongo-instanced-ip-address>:27017"
write_uri = "mongodb://<mongo-instanced-ip-address>:27017"

read_mongo_options = {
    "uri": mongo_uri,
    "database": "mongodbName",
    "collection": "mongodbCollection",
    "username": "mongodbUsername",
    "password": "mongodbPassword",
    "partitioner": "MongoSamplePartitioner",
    "partitionerOptions.partitionSizeMB": "10",
    "partitionerOptions.partitionKey": "_id"}

ssl_mongo_options = {
    "uri": mongo_ssl_uri,
    "database": "mongodbName",
    "collection": "mongodbCollection",
    "ssl": "true",
    "ssl.domain_match": "false"
}

write_mongo_options = {
    "uri": write_uri,
    "database": "mongodbName",
    "collection": "mongodbCollection",
    "username": "mongodbUsername",
    "password": "mongodbPassword",
}

# Get DynamicFrame from MongoDB
dynamic_frame = glueContext.create_dynamic_frame.from_options(connection_type="mongodb",
                                                              connection_options=read_mongo_options)

# Write DynamicFrame to MongoDB
glueContext.write_dynamic_frame.from_options(dynamicFrame, connection_type="mongodb", connection_options=write_mongo_options)

job.commit()

Scala

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 DEFAULT_URI: String = "mongodb://<mongo-instanced-ip-address>:27017"
  val WRITE_URI: String = "mongodb://<mongo-instanced-ip-address>:27017"
  lazy val defaultJsonOption = jsonOptions(DEFAULT_URI)
  lazy val writeJsonOption = jsonOptions(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 MongoDB
    val dynamicFrame: DynamicFrame = glueContext.getSource("mongodb", defaultJsonOption).getDynamicFrame()

    // Write DynamicFrame to MongoDB
    glueContext.getSink("mongodb", writeJsonOption).writeDynamicFrame(dynamicFrame)

    Job.commit()
  }

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

Indicazioni di riferimento alle opzioni di connessione a MongoDB

Designa una connessione a MongoDB. Le opzioni di connessione differiscono per una connessione sorgente e una connessione sink.

Queste proprietà di connessione sono condivise tra le connessioni di origine e sink:

• connectionName: utilizzato per la lettura/scrittura. Il nome di una connessione a MongoDB AWS Glue configurata per fornire informazioni di autenticazione e sulla rete al metodo di connessione. Quando una connessione AWS Glue è configurata come descritto nella sezione precedente, Configurazione delle connessioni MongoDB, la fornitura di connectionName sostituirà la necessità di fornire le opzioni di connessione "uri", "username" e "password".

• "uri": (Obbligatorio) L'host MongoDB da cui leggere, formattato come mongodb://<host>:<port>. Utilizzato nelle versioni AWS Glue precedenti alla versione di AWS Glue 4.0.

• "connection.uri": (Obbligatorio) L'host MongoDB da cui leggere, formattato come mongodb://<host>:<port>. Utilizzato nelle versioni AWS Glue 4.0 e successive.

• "username": (Obbligatorio) Il nome utente MongoDB.

• "password": (Obbligatorio) La password di MongoDB.

• "database": (Obbligatorio) Il database MongoDB da cui leggere. Questa opzione può anche essere trasmessa in additional_options quando si chiama glue_context.create_dynamic_frame_from_catalog nello script di processo.

• "collection": (Obbligatorio) La raccolta MongoDB da cui leggere. Questa opzione può anche essere trasmessa in additional_options quando si chiama glue_context.create_dynamic_frame_from_catalog nello script di processo.

"connectionType": "mongodb" come sorgente

Utilizzare le seguenti opzioni di connessione con "connectionType": "mongodb" come origine:

• "ssl": (Facoltativo) Se il valore è true, avvia una connessione SSL. Il valore predefinito è false.

• "ssl.domain_match": (Facoltativo) Se i valori true e ssl sono true, viene eseguito il controllo della corrispondenza del dominio. Il valore predefinito è true.

• "batchSize": (Facoltativo): il numero di documenti da restituire per ogni batch, utilizzato all'interno del cursore dei batch interni.

• "partitioner": (Facoltativo): il nome della classe del partizionatore per la lettura dei dati di input da MongoDB. Il connettore fornisce i seguenti partizionatori:

  • MongoDefaultPartitioner (impostazione predefinita) (non supportato in AWS Glue 4.0)

  • MongoSamplePartitioner (Richiede MongoDB 3.2 o versioni successive) (non supportato in AWS Glue 4.0)

  • MongoShardedPartitioner (non supportato in AWS Glue 4.0)

  • MongoSplitVectorPartitioner (non supportato in AWS Glue 4.0)

  • MongoPaginateByCountPartitioner (non supportato in AWS Glue 4.0)

  • MongoPaginateBySizePartitioner (non supportato in AWS Glue 4.0)

  • com.mongodb.spark.sql.connector.read.partitioner.SinglePartitionPartitioner

  • com.mongodb.spark.sql.connector.read.partitioner.ShardedPartitioner

  • com.mongodb.spark.sql.connector.read.partitioner.PaginateIntoPartitionsPartitioner

• "partitionerOptions" ( Facoltativo): opzioni per il partizionatore designato. Per ogni partizionatore sono supportate le seguenti opzioni:

  • MongoSamplePartitioner: partitionKey, partitionSizeMB, samplesPerPartition

  • MongoShardedPartitioner: shardkey

  • MongoSplitVectorPartitioner: partitionKey, partitionSizeMB

  • MongoPaginateByCountPartitioner: partitionKey, numberOfPartitions

  • MongoPaginateBySizePartitioner: partitionKey, partitionSizeMB

  Per ulteriori informazioni su queste opzioni, vedere Partitioner Configuration (Configurazione partizionatore) nella documentazione di MongoDB.

"connectionType": "mongodb" come sink

Utilizzare le seguenti opzioni di connessione con "connectionType": "mongodb" come sink:

• "ssl": (Facoltativo) Se il valore è true, avvia una connessione SSL. Il valore predefinito è false.

• "ssl.domain_match": (Facoltativo) Se i valori true e ssl sono true, viene eseguito il controllo della corrispondenza del dominio. Il valore predefinito è true.

• "extendedBsonTypes": (Facoltativo) Se il valore ètrue, permette i tipi BSON estesi durante la scrittura di dati su MongoDB. Il valore predefinito è true.

• "replaceDocument": (Facoltativo) Se il valore è true, sostituisce l'intero documento quando si salvano set di dati che contengono un campo _id. Se il valore è false, vengono aggiornati solo i campi del documento che corrispondono ai campi del set di dati. Il valore predefinito è true.

• "maxBatchSize": (Facoltativo): la dimensione massima del batch per le operazioni in blocco durante il salvataggio dei dati. Il valore predefinito è 512.

• "retryWrites": (Facoltativo): riprova automaticamente alcune operazioni di scrittura una sola volta se AWS Glue rileva un errore di rete.