Tipos de conexión y opciones para ETL en AWS Glue - AWS Glue

Tipos de conexión y opciones para ETL en AWS Glue

En AWS Glue, varios métodos y transformaciones de PySpark y Scala especifican el tipo de conexión utilizando un parámetro connectionType. Especifican las opciones de conexión mediante un parámetro connectionOptions o options.

El parámetro connectionType puede adoptar los valores que se muestran en la tabla siguiente. Los valores del parámetro connectionOptions (u options) asociados para cada tipo se documentan en las secciones siguientes. Salvo que se indique lo contrario, los parámetros se aplican cuando la conexión se utiliza como origen o receptor.

Para obtener un código de muestra que ilustra la configuración y que utiliza las opciones de conexión, consulte Ejemplos: configuración de tipos y opciones de conexión.

connectionType Se conecta a
custom.* Almacenes de datos Spark, Athena o JDBC (consulte Valores de conexiones de tipo personalizada y AWS Marketplace
documentdb Base de datos de Amazon DocumentDB (compatible con MongoDB)
dynamodb Base de datos Amazon DynamoDB
kafka Kafka o Amazon Managed Streaming for Apache Kafka
kinesis Amazon Kinesis Data Streams
marketplace.* Almacenes de datos Spark, Athena o JDBC (consulte Valores de conexiones de tipo personalizada y AWS Marketplace)
mongodb Base de datos de MongoDB
mysql Base de datos de MySQL (consulte Valores connectionType de JDBC)
oracle Base de datos de Oracle (consulte Valores connectionType de JDBC)
orc Archivos almacenados en Amazon Simple Storage Service (Amazon S3) con el formato de archivos Apache Hive Optimized Row Columnar (ORC)
parquet Archivos almacenados en Amazon S3 con el formato de archivos Apache Parquet
postgresql Base de datos de PostgreSQL (consulte Valores connectionType de JDBC)
redshift Base de datos de Amazon Redshift (consulte Valores connectionType de JDBC)
s3 Amazon S3
sqlserver Base de datos de Microsoft SQL Server (consulte Valores connectionType de JDBC)

"connectionType": "documentdb"

Designa una conexión a Amazon DocumentDB (compatible con MongoDB).

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

"connectionType": "documentdb" como origen

Utilice las siguientes opciones de conexión con "connectionType": "documentdb" como origen:

  • "uri": (obligatorio) el host de Amazon DocumentDB del que se va a leer, con formato mongodb://<host>:<port>.

  • "database": (obligatorio) la base de datos de Amazon DocumentDB de la que se va a leer.

  • "collection": (obligatorio) la recopilación de Amazon DocumentDB de la que se va a leer.

  • "username": (obligatorio) nombre de usuario de Amazon DocumentDB.

  • "password": (obligatorio) la contraseña de Amazon DocumentDB.

  • "ssl": (obligatorio si usa SSL) si su conexión usa SSL, debe incluir esta opción con el valor "true".

  • "ssl.domain_match": (obligatorio si usa SSL) si su conexión usa SSL, debe incluir esta opción con el valor "false".

  • "batchSize": (opcional): el número de documentos que se deben devolver por lote, que se utilizan dentro del cursor de lotes internos.

  • "partitioner": (opcional): el nombre de la clase del particionador para leer los datos de entrada de Amazon DocumentDB. El conector proporciona los siguientes particionadores:

    • MongoDefaultPartitioner (predeterminado)

    • MongoSamplePartitioner

    • MongoShardedPartitioner

    • MongoSplitVectorPartitioner

    • MongoPaginateByCountPartitioner

    • MongoPaginateBySizePartitioner

  • "partitionerOptions" ( opcional): opciones para el particionador designado. Se admiten las siguientes opciones para cada particionador:

    • MongoSamplePartitioner: partitionKey, partitionSizeMB, samplesPerPartition

    • MongoShardedPartitioner: shardkey

    • MongoSplitVectorPartitioner: partitionKey, partitionSizeMB

    • MongoPaginateByCountPartitioner: partitionKey, numberOfPartitions

    • MongoPaginateBySizePartitioner: partitionKey, partitionSizeMB

    Para obtener más información acerca de estas opciones, consulte Partitioner Configuration en la documentación de MongoDB. Para ver el código de muestra, consulte Ejemplos: configuración de tipos y opciones de conexión.

"connectionType": "documentdb" como receptor

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

  • "uri": (obligatorio) el host de Amazon DocumentDB al que se va a escribir, con formato mongodb://<host>:<port>.

  • "database": (obligatorio) la base de datos de Amazon DocumentDB a la que se va a escribir.

  • "collection": (obligatorio) la recopilación de Amazon DocumentDB a la que se va a escribir.

  • "username": (obligatorio) nombre de usuario de Amazon DocumentDB.

  • "password": (obligatorio) la contraseña de Amazon DocumentDB.

  • "extendedBsonTypes": (opcional) si se establece en true, permite los tipos de BSON extendidos al escribir datos en Amazon DocumentDB. El valor predeterminado es true.

  • "replaceDocument": (opcional) si es true, reemplaza todo el documento al guardar conjuntos de datos que contienen un campo _id. Si es false, solo se actualizan los campos del documento que coinciden con los campos del conjunto de datos. El valor predeterminado es true.

  • "maxBatchSize": (opcional): el tamaño máximo del lote para operaciones en bloque al guardar datos. El valor predeterminado es 512.

Para ver el código de muestra, consulte Ejemplos: configuración de tipos y opciones de conexión.

"connectionType": "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:

  • "dynamodb.input.tableName": (obligatorio) tabla de DynamoDB desde la que se debe leer.

  • "dynamodb.throughput.read.percent": (opcional) porcentaje de unidades de capacidad de lectura (RCU) que se usará. El valor predeterminado se establece en "0,5". Los valores aceptables abarcan de "0,1" a "1,5", inclusive.

    • 0.5 representa la tasa de lectura predeterminada, es decir que AWS Glue intentará consumir la mitad de la capacidad de lectura de la tabla. Si usted aumenta el valor por encima de 0.5, AWS Glue incrementará la tasa de solicitudes; si reduce el valor por debajo de 0.5, disminuirá la tasa de solicitudes de lectura. (La tasa de lectura real varía, según diversos factores tales como el hecho de que exista o no una distribución uniforme de claves en la tabla de DynamoDB).

    • Cuando la tabla de DynamoDB está en modo bajo demanda, AWS Glue maneja la capacidad de lectura de la tabla como 40 000. Para exportar una tabla de gran extensión, recomendamos cambiar la tabla de DynamoDB al modo bajo demanda.

  • "dynamodb.splits": (opcional) define la cantidad de particiones en las que dividimos esta tabla de DynamoDB al leerla. El valor predeterminado se establece en "1". Los valores aceptables abarcan de "1" a "1,000,000", inclusive.

    • 1 representa que no hay paralelismo. Se recomienda especialmente que especifique un valor mayor para un mejor rendimiento mediante la fórmula siguiente.

    • Le recomendamos que calcule numSlots con la siguiente fórmula, y lo use como dynamodb.splits. Si necesita más rendimiento, le recomendamos que amplíe su trabajo al aumentar el número de DPU.

      • numExecutors =

        • (DPU - 1) * 2 - 1 si WorkerType es Standard

        • (NumberOfWorkers - 1) si WorkerType es G.1X o G.2X

      • numSlotsPerExecutor =

        • 4 si WorkerType es Standard

        • 8 si WorkerType es G.1X

        • 16 si WorkerType es G.2X

      • numSlots = numSlotsPerExecutor * numExecutors

  • "dynamodb.sts.roleArn": (opcional) el ARN de rol de IAM que se asumirá para el acceso entre cuentas. Este parámetro se encuentra disponible en AWS Glue 1.0 o posterior.

  • "dynamodb.sts.roleSessionName": (opcional) nombre de sesión STS. El valor predeterminado se establece en “glue-dynamodb-read-sts-session”. Este parámetro se encuentra disponible en AWS Glue 1.0 o posterior.

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

AWS Glue admite la lectura de datos de tablas de DynamoDB de otra cuenta de AWS. Para obtener más información, consulte Acceso entre cuentas y entre regiones a tablas de DynamoDB .

nota

El lector de ETL de DynamoDB no es compatible con filtros ni predicados de inserción.

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

Esta característica está en versión preliminar para AWS Glue y está sujeta a cambios.

Además del conector de ETL de DynamoDB de AWS Glue, AWS Glue ofrece un conector de exportación de DynamoDB, que invoca una solicitud ExportTableToPointInTime de DynamoDB y la almacena en una ubicación de Simple Storage Service (Amazon S3) que se proporcione, con el formato de JSON de DynamoDB. Después, AWS Glue crea un objeto DynamicFrame leyendo los datos desde la ubicación de exportación de Simple Storage Service (Amazon S3).

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.

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:

  • "dynamodb.export": (obligatorio) valor de cadena:

    • Si se configura como ddb, habilita el conector de exportación de DynamoDB de AWS Glue cuando se invoque una nueva ExportTableToPointInTimeRequest durante el trabajo de AWS Glue. Se generará una nueva exportación con la ubicación pasada desde dynamodb.s3.bucket y dynamodb.s3.prefix.

    • Si se configura como s3, habilita el conector de exportación de DynamoDB de AWS Glue, pero omite la creación de una nueva exportación de DynamoDB y, en su lugar, utiliza dynamodb.s3.bucket y dynamodb.s3.prefix como ubicación de Simple Storage Service (Amazon S3) de una exportación anterior de esa tabla.

  • "dynamodb.tableArn": (obligatorio) tabla de DynamoDB desde la que se debe leer.

  • "dynamodb.unnestDDBJson": (opcional) toma un valor booleano. Si se configura como true, realiza una transformación no anidada de la estructura JSON de DynamoDB que está presente en las exportaciones. El valor predeterminado está configurado como false.

  • "dynamodb.s3.bucket": (opcional) indica la ubicación del bucket de Simple Storage Service (Amazon S3) en el que debe llevarse a cabo el proceso ExportTableToPointInTime de DynamoDB. El formato de archivo para la exportación es DynamoDB JSON.

    • "dynamodb.s3.prefix": (opcional) indica la ubicación del prefijo de Simple Storage Service (Amazon S3) dentro del bucket de Simple Storage Service (Amazon S3) en el que se almacenarán las cargas ExportTableToPointInTime de DynamoDB. Si no se especifica dynamodb.s3.prefix ni dynamodb.s3.bucket, estos valores adoptarán de manera predeterminada la ubicación del directorio temporal especificada en la configuración del trabajo de AWS Glue. Para obtener más información, consulte Parámetros especiales utilizados por AWS Glue.

    • "dynamodb.s3.bucketOwner": indica el propietario del bucket que se necesita para el acceso entre cuentas de Simple Storage Service (Amazon S3).

  • "dynamodb.sts.roleArn": (opcional) el ARN del rol de IAM que se asumirá para el acceso entre cuentas o el acceso entre regiones para la tabla de DynamoDB. Nota: El mismo ARN del rol de IAM se utilizará para acceder a la ubicación de Simple Storage Service (Amazon S3) especificada para la solicitud ExportTableToPointInTime.

  • "dynamodb.sts.roleSessionName": (opcional) nombre de sesión STS. El valor predeterminado se establece en “glue-dynamodb-read-sts-session”.

nota

DynamoDB tiene requisitos específicos para invocar las solicitudes ExportTableToPointInTime. Para obtener más información, consulte Solicitud de una exportación de tabla en DynamoDB. Por ejemplo, debe estar habilitada la restauración a un momento dado (PITR) en la tabla para utilizar este conector. El conector de DynamoDB también es compatible con el cifrado de KMS para las exportaciones de DynamoDB a Simple Storage Service (Amazon S3). Proporcionar la configuración de seguridad en la configuración del trabajo de AWS Glue habilita el cifrado de KMS para una exportación de DynamoDB. La clave de KMS debe estar en la misma región que el bucket de Simple Storage Service (Amazon S3).

Tenga en cuenta que se aplican cargos adicionales por la exportación de DynamoDB y los costos de almacenamiento de Simple Storage Service (Amazon S3). Los datos exportados en Simple Storage Service (Amazon S3) se mantienen cuando finaliza la ejecución de un trabajo para que pueda reutilizarlos sin exportaciones adicionales de DynamoDB. Un requisito para utilizar este conector es que la recuperación a un momento dado (PITR) esté habilitada para la tabla.

El conector de ETL y el conector de exportación de DynamoDB no son compatibles con la aplicación de filtros o predicados de inserción en el origen 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>", } ) 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()) )

Examen de la estructura JSON de DynamoDB

Las exportaciones de DynamoDB con el conector de exportación de DynamoDB de AWS Glue pueden generar 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; la primera es un indicador booleano que se pasa con el conector de exportación de DynamoDB de AWS Glue. La segunda consiste en llamar a la propia función de transformación.

Los siguientes ejemplos de código muestran cómo utilizar el conector de exportación de DynamoDB de AWS Glue, invocar un desanidamiento e imprimir el número de particiones:

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.unnestDDBJson": True, "dynamodb.s3.bucket": "<bucket name>", "dynamodb.s3.prefix": "<bucket prefix>", "dynamodb.s3.bucketOwner": "<account_id>", } ) 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.unnestDDBJson" -> true "dynamodb.s3.bucket" -> "<bucket name>", "dynamodb.s3.prefix" -> "<bucket prefix>", "dynamodb.s3.bucketOwner" -> "<account_id of bucket>", )) ).getDynamicFrame() print(dynamicFrame.getNumPartitions()) Job.commit() } }

La otra invocación de la transformación se realiza mediante una llamada a la función DynamicFrame independiente. Para obtener más información, consulte Clase DynamicFrame para Python y Clase DynamicFrame de Scala de AWS Glue para Scala.

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

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

  • "dynamodb.output.tableName": (obligatorio) tabla de DynamoDB en la que se va a escribir.

  • "dynamodb.throughput.write.percent": (opcional) porcentaje de unidades de capacidad de escritura (WCU) que se usará. El valor predeterminado se establece en "0,5". Los valores aceptables abarcan de "0,1" a "1,5", inclusive.

    • 0.5 representa la tasa de escritura predeterminada, es decir que AWS Glue intentará consumir la mitad de la capacidad de escritura de la tabla. Si usted aumenta el valor por encima de 0,5, AWS Glue incrementará la tasa de solicitudes; si reduce el valor por debajo de 0,5, disminuirá la tasa de solicitudes de escritura. (La tasa de escritura real varía en función de diversos factores, tales como el hecho de que exista o no una distribución uniforme de claves en la tabla de DynamoDB).

    • Cuando la tabla de DynamoDB está en modo bajo demanda, AWS Glue maneja la capacidad de escritura de la tabla como 40000. Para importar una tabla grande, recomendamos cambiar la tabla de DynamoDB al modo bajo demanda.

  • "dynamodb.output.numParallelTasks": (opcional) define el número de tareas paralelas que escriben en DynamoDB al mismo tiempo. Se utiliza para calcular WCU permisiva por tarea de Spark. Si no desea controlar estos detalles, no es necesario que especifique este parámetro.

    • permissiveWcuPerTask = TableWCU * dynamodb.throughput.write.percent / dynamodb.output.numParallelTasks

    • Si no especifica este parámetro, la WCU permisiva por tarea de Spark se calculará automáticamente mediante la siguiente fórmula:

      • numPartitions = dynamicframe.getNumPartitions()

      • numExecutors =

        • (DPU - 1) * 2 - 1 si WorkerType es Standard

        • (NumberOfWorkers - 1) si WorkerType es G.1X o G.2X

      • numSlotsPerExecutor =

        • 4 si WorkerType es Standard

        • 8 si WorkerType es G.1X

        • 16 si WorkerType es G.2X

      • numSlots = numSlotsPerExecutor * numExecutors

      • numParallelTasks = min(numPartitions, numSlots)

    • Ejemplo 1. DPU=10, tipo de empleado=estándar. La entrada DynamicFrame tiene 100 particiones RDD.

      • numPartitions = 100

      • numExecutors = (10 - 1) * 2 - 1 = 17

      • numSlots = 4 * 17 = 68

      • numParallelTasks = min(100, 68) = 68

    • Ejemplo 2. DPU=10, tipo de empleado=estándar. La entrada DynamicFrame tiene 20 particiones RDD.

      • numPartitions = 20

      • numExecutors = (10 - 1) * 2 - 1 = 17

      • numSlots = 4 * 17 = 68

      • numParallelTasks = min(20, 68) = 20

  • "dynamodb.output.retry": (opcional) define el número de reintentos que realizamos cuando hay una ProvisionedThroughputExceededException de DynamoDB. El valor predeterminado se establece en “10”.

  • "dynamodb.sts.roleArn": (opcional) el ARN de rol de IAM que se asumirá para el acceso entre cuentas.

  • "dynamodb.sts.roleSessionName": (opcional) nombre de sesión STS. El valor predeterminado se establece en “glue-dynamodb-write-sts-session”.

nota

La versión de AWS Glue 1.0 o posterior admite la escritura de DynamoDB.

nota

AWS Glue admite la escritura de datos de tablas de DynamoDB en otra cuenta de AWS. Para obtener más información, consulte Acceso entre cuentas y entre regiones a tablas de DynamoDB .

En los siguientes ejemplos de código, se muestra cómo leer 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() } }

“connectionType”: “kafka”

Designa una conexión a un clúster de Kafka o a un clúster de Amazon Managed Streaming for Apache Kafka.

Puede utilizar los métodos siguientes en el objeto GlueContext para consumir registros de un origen de streaming de Kafka:

  • getCatalogSource

  • getSource

  • getSourceWithFormat

Si utiliza getCatalogSource, el trabajo tiene la base de datos del Data Catalog y la información del nombre de la tabla, y puede usarla para obtener algunos parámetros básicos para la lectura desde el flujo de Apache Kafka. Si utiliza getSource, debe especificar estos parámetros de manera explícita:

Puede especificar estas opciones mediante connectionOptions con GetSource, options con getSourceWithFormat, o additionalOptions con getCatalogSource.

Utilice las siguientes opciones de conexión con "connectionType": "kafka":

  • bootstrap.servers (obligatorio): una lista de direcciones URL de servidor de arranque, por ejemplo, como b-1.vpc-test-2.o4q88o.c6.kafka.us-east-1.amazonaws.com:9094. Esta opción debe especificarse en la llamada a la API o definirse en los metadatos de la tabla en el Data Catalog.

  • security.protocol (obligatorio): el protocolo que se utiliza para la comunicación con los agentes. Los valores posibles son "SSL" o "PLAINTEXT".

  • topicName (obligatorio), nombre del tema como se especifica en Apache Kafka. Debe especificar al menos una opción entre "topicName", "assign" o "subscribePattern".

  • "assign": (obligatorio) las TopicPartitions específicas que se consumirán. Debe especificar al menos una opción entre "topicName", "assign" o "subscribePattern".

  • "subscribePattern": (obligatorio) cadena de expresiones regulares de Java que identifica la lista de temas a la que desea suscribirse. Debe especificar al menos una opción entre "topicName", "assign" o "subscribePattern".

  • classification (opcional)

  • delimiter (opcional)

  • "startingOffsets": (opcional) posición inicial en el tema de Kafka para leer los datos. Los valores posibles son "earliest" o "latest". El valor predeterminado es "latest".

  • "endingOffsets": (opcional) el punto final cuando finaliza una consulta por lotes. Los valores posibles son "latest" o una cadena JSON que especifica una compensación final para cada TopicPartition.

    Para la cadena JSON, el formato es {"topicA":{"0":23,"1":-1},"topicB":{"0":-1}}. El valor -1 como compensación representa "latest".

  • "pollTimeoutMs": (opcional) tiempo de espera en milisegundos para sondear datos de Kafka en ejecutores de trabajos de Spark. El valor predeterminado es 512.

  • "numRetries": (opcional) el número de veces que se reintenta antes de no obtener las compensaciones de Kafka. El valor predeterminado es 3.

  • "retryIntervalMs": (opcional) tiempo en milisegundos para esperar antes de volver a intentar obtener compensaciones Kafka. El valor predeterminado es 10.

  • "maxOffsetsPerTrigger": (opcional) el límite de velocidad en el número máximo de compensaciones que se procesan por intervalo de desencadenador. El número total de compensaciones especificado se divide de forma proporcional entre topicPartitions de diferentes volúmenes. El valor predeterminado es nulo, lo que significa que el consumidor lee todos las compensaciones hasta la última compensación conocida.

  • "minPartitions": (opcional) el número mínimo deseado de particiones para leer desde Kafka. El valor predeterminado es nulo, lo que significa que el número de particiones de Spark es igual al número de particiones de Kafka.

  • "includeHeaders": (opcional) si se deben incluir los encabezados de Kafka. Cuando la opción se establece en “verdadero”, la salida de datos contendrá una columna adicional denominada “glue_streaming_kafka_headers” con el tipo Array[Struct(key: String, value: String)]. El valor predeterminado es “falso”. Esta opción solo se encuentra disponible en la versión 3.0 de AWS Glue.

“connectionType”: “kinesis”

Designa opciones de conexión para Amazon Kinesis Data Streams.

Puede leer desde una secuencia de datos de Amazon Kinesis mediante la información almacenada en una tabla de Data Catalog o al proporcionar información para acceder directamente a la secuencia de datos. Si accede directamente a la secuencia de datos, utilice estas opciones para proporcionar información sobre cómo acceder a la secuencia de datos.

Si utiliza getCatalogSource o create_data_frame_from_catalog para consumir registros de una fuente de streaming de Kinesis, el trabajo tiene la base de datos de Data Catalog y la información del nombre de la tabla, y puede utilizarla para obtener algunos parámetros básicos para la lectura de la fuente de streaming de Kinesis. Si utiliza getSource, getSourceWithFormat o create_data_frame_from_options, debe especificar estos parámetros básicos mediante las opciones de conexión descritas aquí.

Puede especificar las opciones de conexión para Kinesis al utilizar los siguientes argumentos para los métodos especificados en la clase GlueContext.

  • Scala

    • connectionOptions: se debe utilizar con getSource

    • additionalOptions: se debe utilizar con getCatalogSource

    • options: se debe utilizar con getSourceWithFormat

  • Python

    • connection_options: se debe utilizar con create_data_frame_from_options

    • additional_options: se debe utilizar con create_data_frame_from_catalog

    • options: se debe utilizar con getSource

Utilice las siguientes opciones de conexión para los orígenes de datos de streaming de Kinesis:

  • streamARN (obligatorio) el ARN de flujo de datos de Kinesis.

  • classification (opcional)

  • delimiter (opcional)

  • "startingPosition": (opcional) posición inicial en el flujo de datos de Kinesis para leer los datos. Los valores posibles son "latest", "trim_horizon", o "earliest". El valor predeterminado es "latest".

  • "awsSTSRoleARN": (opcional) el nombre de recurso de Amazon (ARN) del rol de que se asumirá mediante AWS Security Token Service (AWS STS). Este rol debe tener permisos para describir o leer operaciones de registros de la secuencia de datos de Kinesis. Debe utilizar este parámetro para acceder a un flujo de datos de otra cuenta. Se utiliza junto con "awsSTSSessionName".

  • "awsSTSSessionName": (opcional) identificador de la sesión que asume el rol mediante AWS STS. Debe utilizar este parámetro para acceder a una secuencia de datos de otra cuenta. Se utiliza junto con "awsSTSRoleARN".

  • "maxFetchTimeInMs": (opcional) el tiempo máximo empleado en el ejecutor de trabajos para obtener un registro del flujo de datos de Kinesis por fragmento, especificado en milisegundos (ms). El valor predeterminado es 1000.

  • "maxFetchRecordsPerShard": (opcional) número máximo de registros que se recuperará por fragmento en el flujo de datos de Kinesis. El valor predeterminado es 100000.

  • "maxRecordPerRead": (opcional) número máximo de registros que se recuperará del flujo de datos de Kinesis en cada operación getRecords. El valor predeterminado es 10000.

  • "addIdleTimeBetweenReads": (opcional) Agrega un retardo de tiempo entre dos operaciones getRecords consecutivas. El valor predeterminado es "False". Esta opción sólo se puede configurar para Glue versión 2.0 y superior.

  • "idleTimeBetweenReadsInMs": (opcional) el tiempo mínimo de retraso entre dos operaciones getRecords consecutivas, especificado en ms. El valor predeterminado es 1000. Esta opción sólo se puede configurar para Glue versión 2.0 y superior.

  • "describeShardInterval": (opcional) el intervalo mínimo de tiempo entre dos llamadas a la API ListShards para que su script considere cambios en los fragmentos. Para obtener más información, consulte Estrategias para cambios en los fragmentos en la Guía del desarrollador de Amazon Kinesis Data Streams. El valor predeterminado es 1s.

  • "numRetries": (opcional) el número máximo de reintentos para las solicitudes de la API de Kinesis Data Streams. El valor predeterminado es 3.

  • "retryIntervalMs": (opcional) el período de tiempo de enfriamiento (especificado en ms) antes de volver a intentar la llamada a la API de Kinesis Data Streams. El valor predeterminado es 1000.

  • "maxRetryIntervalMs": (opcional) el período de tiempo de enfriamiento máximo (especificado en ms) entre dos intentos de llamada a la API de Kinesis Data Streams. El valor predeterminado es 10000.

  • "avoidEmptyBatches": (opcional) evita crear un trabajo de microlotes vacío al comprobar si hay datos no leídos en el flujo de datos de Kinesis antes de que se inicie el lote. El valor predeterminado es "False".

"connectionType": "mongodb"

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

"connectionType": "mongodb" como origen

Utilice las siguientes opciones de conexión con "connectionType": "mongodb" como origen:

  • "uri": (obligatorio) el host de MongoDB del que se va a leer, con formato mongodb://<host>:<port>.

  • "database": (obligatorio) la base de datos de MongoDB de la que se va a leer. Esta opción también se puede transferir a additional_options al llamar a glue_context.create_dynamic_frame_from_catalog en su script de trabajo.

  • "collection": (obligatorio) la colección de MongoDB de la que se va a leer. Esta opción también se puede transferir a additional_options al llamar a glue_context.create_dynamic_frame_from_catalog en su script de trabajo.

  • "username": (obligatorio) el nombre de usuario de MongoDB.

  • "password": (obligatorio) la contraseña de MongoDB.

  • "ssl": (opcional) si es true, inicia una conexión SSL. El valor predeterminado es false.

  • "ssl.domain_match": (opcional) si es true y ssl es true, se realiza la comprobación de coincidencia de dominio. El valor predeterminado es true.

  • "batchSize": (opcional): el número de documentos que se deben devolver por lote, que se utilizan dentro del cursor de lotes internos.

  • "partitioner": (opcional): el nombre de la clase del particionador para leer los datos de entrada de MongoDB. El conector proporciona los siguientes particionadores:

    • MongoDefaultPartitioner (predeterminado)

    • MongoSamplePartitioner (Requiere MongoDB 3.2 o posterior)

    • MongoShardedPartitioner

    • MongoSplitVectorPartitioner

    • MongoPaginateByCountPartitioner

    • MongoPaginateBySizePartitioner

  • "partitionerOptions" ( opcional): opciones para el particionador designado. Se admiten las siguientes opciones para cada particionador:

    • MongoSamplePartitioner: partitionKey, partitionSizeMB, samplesPerPartition

    • MongoShardedPartitioner: shardkey

    • MongoSplitVectorPartitioner: partitionKey, partitionSizeMB

    • MongoPaginateByCountPartitioner: partitionKey, numberOfPartitions

    • MongoPaginateBySizePartitioner: partitionKey, partitionSizeMB

    Para obtener más información acerca de estas opciones, consulte Partitioner Configuration en la documentación de MongoDB. Para ver el código de muestra, consulte Ejemplos: configuración de tipos y opciones de conexión.

"connectionType": "mongodb" como receptor

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

  • "uri": (obligatorio) el host de MongoDB en el que se va a escribir, con formato mongodb://<host>:<port>.

  • "database": (obligatorio) la base de datos de MongoDB en la que se va a escribir.

  • "collection": (obligatorio) la colección de MongoDB en la que se va a escribir.

  • "username": (obligatorio) el nombre de usuario de MongoDB.

  • "password": (obligatorio) la contraseña de MongoDB.

  • "ssl": (opcional) si es true, inicia una conexión SSL. El valor predeterminado es false.

  • "ssl.domain_match": (opcional) si es true y ssl es true, se realiza la comprobación de coincidencia de dominio. El valor predeterminado es true.

  • "extendedBsonTypes": (opcional) si es true, habilitan los tipos de BSON ampliados al escribir datos en MongoDB. El valor predeterminado es true.

  • "replaceDocument": (opcional) si es true, reemplaza todo el documento al guardar conjuntos de datos que contienen un campo _id. Si es false, solo se actualizan los campos del documento que coinciden con los campos del conjunto de datos. El valor predeterminado es true.

  • "maxBatchSize": (opcional): el tamaño máximo del lote para operaciones en bloque al guardar datos. El valor predeterminado es 512.

Para ver el código de muestra, consulte Ejemplos: configuración de tipos y opciones de conexión.

"connectionType": "orc"

Designa una conexión a archivos almacenados en Amazon S3 en formato de archivo Apache Hive Optimized Row Columnar (ORC).

Utilice las siguientes opciones de conexión con "connectionType": "orc":

  • paths: (obligatorio) una lista de las rutas de Amazon S3 desde las que se va a leer.

  • (Otros pares de nombre/valor de la opción): cualquier opción adicional, incluidas las opciones de formato, se pasa directamente a SparkSQL DataSource. Para obtener más información, consulte el artículo acerca del origen de datos Redshift de Spark.

"connectionType": "parquet"

Designa una conexión a archivos almacenados en Amazon S3 con el formato de archivo Apache Parquet.

Utilice las siguientes opciones de conexión con "connectionType": "parquet":

  • paths: (obligatorio) una lista de las rutas de Amazon S3 desde las que se va a leer.

  • (Otros pares de nombre/valor de la opción): cualquier opción adicional, incluidas las opciones de formato, se pasa directamente a SparkSQL DataSource. Para obtener más información, consulte Amazon Redshift data source for Spark en el sitio web de GitHub.

"connectionType": "s3"

Designa una conexión a Amazon S3.

Utilice las siguientes opciones de conexión con "connectionType": "s3":

  • "paths": (obligatorio) una lista de las rutas de Amazon S3 desde las que se va a leer.

  • "exclusions": (opcional) cadena que contiene una lista JSON de patrones glob de estilo Unix para excluir. Por ejemplo, "[\"**.pdf\"]" excluye todos los archivos PDF. Para obtener más información acerca de la sintaxis glob que admite AWS Glue, consulte Incluir y excluir patrones.

  • "compressionType": o "compression": (opcional) especifica la forma en que los datos se comprimen. Use "compressionType" para orígenes de Amazon S3 y "compression" para destinos de Amazon S3. Po lo general no es necesario si los datos tienen una extensión de archivo estándar. Los posibles valores son "gzip" y "bzip").

  • "groupFiles": (opcional) la agrupación de archivos se habilita de forma predeterminada cuando la entrada contiene más de 50 000 archivos. Para habilitar las agrupaciones con menos de 50 000 archivos, establezca este parámetro en "inPartition". Para deshabilitar las agrupaciones con más de 50 000 archivos, establezca este parámetro en "none".

  • "groupSize": (opcional) tamaño del grupo de destino en bytes. El valor predeterminado se calcula en función del tamaño de los datos de entrada y el tamaño de su clúster. Cuando hay menos de 50 000 archivos de entrada, "groupFiles" debe establecerse en "inPartition" para que este valor surta efecto.

  • "recurse": (opcional) si se establece en verdadero, lee recursivamente archivos en todos los subdirectorios de las rutas especificadas.

  • "maxBand": (opcional, avanzada) esta opción controla la duración en milisegundos después de la que es probable que el listado de s3 sea coherente. Se realiza un seguimiento de los archivos cuyas marcas de tiempo de modificación estén comprendidas en los últimos milisegundos de maxBand, en especial cuando se utilizan JobBookmarks para obtener coherencia eventual de Amazon S3. La mayoría de los usuarios no tienen que establecer esta opción. El valor predeterminado es 900 000 milisegundos, o 15 minutos.

  • "maxFilesInBand": (opcional, avanzada) esta opción especifica el número máximo de archivos que deben guardarse desde los últimos segundos de maxBand. Si se supera este número, los archivos adicionales se omiten y solo se procesarán en la siguiente ejecución del flujo de trabajo. La mayoría de los usuarios no tienen que establecer esta opción.

  • "isFailFast": (opcional) esta opción determina si un trabajo de ETL de AWS Glue arroja excepciones de análisis del lector. Si se establece en true, los trabajos fallan rápidamente si cuatro reintentos de la tarea Spark no pueden analizar los datos en forma correcta.

Valores connectionType de JDBC

Esto incluye lo siguiente:

  • "connectionType": "sqlserver": designa una conexión a una base de datos Microsoft SQL Server.

  • "connectionType": "mysql": designa una conexión a una base de datos MySQL.

  • "connectionType": "oracle": designa una conexión a una base de datos Oracle.

  • "connectionType": "postgresql": designa una conexión a una base de datos PostgreSQL.

  • "connectionType": "redshift": designa una conexión a una base de datos de Amazon Redshift.

En la siguiente tabla, se muestran las versiones del controlador JDBC que AWS Glue admite.

Producto Versiones del controlador JDBC para Glue 0.9, 1.0, 2.0 Versiones del controlador JDBC para Glue 3.0
Microsoft SQL Server 6.x 7.x
MySQL 5.1 8.0.23
Oracle Database 11.2 21.1
PostgreSQL 42.1.x 42.2.18
MongoDB 2.0.0 4.0.0
Amazon Redshift 4.1 4.1

Utilice estas opciones con las conexiones JDBC:

  • "url": (obligatorio) la URL de JDBC de la base de datos.

  • "dbtable": la tabla de base de datos donde se efectúa la lectura. Para almacenes de datos de JDBC que admiten esquemas dentro de una base de datos, especifique schema.table-name. Si no se ha proporcionado un esquema, se usa el esquema "public" predeterminado.

  • "redshiftTmpDir": (obligatoria para Amazon Redshift, opcional para otros tipos JDBC) la ruta de Amazon S3 donde se pueden almacenar datos temporales al copiar fuera de la base de datos.

  • "user": (obligatorio) nombre de usuario que se va a utilizar para establecer la conexión.

  • "password": (obligatorio) contraseña que se utilizará para establecer la conexión.

  • (Opcional) las siguientes opciones le permiten proporcionar un controlador JDBC personalizado. Utilice estas opciones si debe utilizar un controlador que AWS Glue no admita de forma nativa. Los trabajos de ETL pueden utilizar diferentes versiones de controladores JDBC para el origen y el destino de los datos, incluso aunque el origen y el destino sean el mismo producto de base de datos. Esto le permite migrar datos entre bases de datos de origen y destino con diferentes versiones. Para usar estas opciones, primero debe cargar el archivo jar del controlador JDBC en Amazon S3.

    • "customJdbcDriverS3Path": ruta de Amazon S3 del controlador JDBC personalizado.

    • "customJdbcDriverClassName": nombre de clase del controlador JDBC.

  • "bulksize": (opcional) se utiliza para configurar inserciones paralelas para acelerar las cargas masivas en los objetivos de JDBC. Especifique un valor entero para el grado de paralelismo que se utilizará al escribir o insertar datos. Esta opción resulta útil para mejorar el rendimiento de las escrituras en bases de datos como Arch User Repository (AUR).

Todos los demás pares nombre /valor que estén presentes en las opciones de una conexión JDBC, incluidas las opciones de formato, se pasarán directamente al origen de datos de SparkSQL subyacente. Para obtener más información, consulte el artículo acerca del origen de datos Redshift de Spark.

En los ejemplos de código siguientes, se muestra cómo se utilizan controladores JDBC personalizados para leer bases de datos JDBC y escribir en ellas. En ellos, se muestra cómo se lee una versión de un producto de base de datos y se escribe en una versión posterior del mismo producto.

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 from pyspark.sql.types import StructType, StructField, IntegerType, StringType sc = SparkContext() glueContext = GlueContext(sc) spark = glueContext.spark_session # Construct JDBC connection options connection_mysql5_options = { "url": "jdbc:mysql://<jdbc-host-name>:3306/db", "dbtable": "test", "user": "admin", "password": "pwd"} connection_mysql8_options = { "url": "jdbc:mysql://<jdbc-host-name>:3306/db", "dbtable": "test", "user": "admin", "password": "pwd", "customJdbcDriverS3Path": "s3://path/mysql-connector-java-8.0.17.jar", "customJdbcDriverClassName": "com.mysql.cj.jdbc.Driver"} connection_oracle11_options = { "url": "jdbc:oracle:thin:@//<jdbc-host-name>:1521/ORCL", "dbtable": "test", "user": "admin", "password": "pwd"} connection_oracle18_options = { "url": "jdbc:oracle:thin:@//<jdbc-host-name>:1521/ORCL", "dbtable": "test", "user": "admin", "password": "pwd", "customJdbcDriverS3Path": "s3://path/ojdbc10.jar", "customJdbcDriverClassName": "oracle.jdbc.OracleDriver"} # Read from JDBC databases with custom driver df_mysql8 = glueContext.create_dynamic_frame.from_options(connection_type="mysql", connection_options=connection_mysql8_options) # Read DynamicFrame from MySQL 5 and write to MySQL 8 df_mysql5 = glueContext.create_dynamic_frame.from_options(connection_type="mysql", connection_options=connection_mysql5_options) glueContext.write_from_options(frame_or_dfc=df_mysql5, connection_type="mysql", connection_options=connection_mysql8_options) # Read DynamicFrame from Oracle 11 and write to Oracle 18 df_oracle11 = glueContext.create_dynamic_frame.from_options(connection_type="oracle", connection_options=connection_oracle11_options) glueContext.write_from_options(frame_or_dfc=df_oracle11, connection_type="oracle", connection_options=connection_oracle18_options)
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 MYSQL_5_URI: String = "jdbc:mysql://<jdbc-host-name>:3306/db" val MYSQL_8_URI: String = "jdbc:mysql://<jdbc-host-name>:3306/db" val ORACLE_11_URI: String = "jdbc:oracle:thin:@//<jdbc-host-name>:1521/ORCL" val ORACLE_18_URI: String = "jdbc:oracle:thin:@//<jdbc-host-name>:1521/ORCL" // Construct JDBC connection options lazy val mysql5JsonOption = jsonOptions(MYSQL_5_URI) lazy val mysql8JsonOption = customJDBCDriverJsonOptions(MYSQL_8_URI, "s3://path/mysql-connector-java-8.0.17.jar", "com.mysql.cj.jdbc.Driver") lazy val oracle11JsonOption = jsonOptions(ORACLE_11_URI) lazy val oracle18JsonOption = customJDBCDriverJsonOptions(ORACLE_18_URI, "s3://path/ojdbc10.jar", "oracle.jdbc.OracleDriver") 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) // Read from JDBC database with custom driver val df_mysql8: DynamicFrame = glueContext.getSource("mysql", mysql8JsonOption).getDynamicFrame() // Read DynamicFrame from MySQL 5 and write to MySQL 8 val df_mysql5: DynamicFrame = glueContext.getSource("mysql", mysql5JsonOption).getDynamicFrame() glueContext.getSink("mysql", mysql8JsonOption).writeDynamicFrame(df_mysql5) // Read DynamicFrame from Oracle 11 and write to Oracle 18 val df_oracle11: DynamicFrame = glueContext.getSource("oracle", oracle11JsonOption).getDynamicFrame() glueContext.getSink("oracle", oracle18JsonOption).writeDynamicFrame(df_oracle11) Job.commit() } private def jsonOptions(uri: String): JsonOptions = { new JsonOptions( s"""{"url": "${url}", |"dbtable":"test", |"user": "admin", |"password": "pwd"}""".stripMargin) } private def customJDBCDriverJsonOptions(uri: String, customJdbcDriverS3Path: String, customJdbcDriverClassName: String): JsonOptions = { new JsonOptions( s"""{"url": "${url}", |"dbtable":"test", |"user": "admin", |"password": "pwd", |"customJdbcDriverS3Path": "${customJdbcDriverS3Path}", |"customJdbcDriverClassName" : "${customJdbcDriverClassName}"}""".stripMargin) } }

Valores de conexiones de tipo personalizada y AWS Marketplace

Esto incluye lo siguiente:

  • "connectionType": "marketplace.athena": designa una conexión a un almacén de datos de Amazon Athena. La conexión utiliza un conector de AWS Marketplace.

  • "connectionType": "marketplace.spark": designa una conexión a un almacén de datos de Apache Spark. La conexión utiliza un conector de AWS Marketplace.

  • "connectionType": "marketplace.jdbc": designa una conexión a un almacén de datos de JDBC. La conexión utiliza un conector de AWS Marketplace.

  • "connectionType": "custom.athena": designa una conexión a un almacén de datos de Amazon Athena. La conexión utiliza un conector personalizado que se carga a AWS Glue Studio.

  • "connectionType": "custom.spark": designa una conexión a un almacén de datos de Apache Spark. La conexión utiliza un conector personalizado que se carga a AWS Glue Studio.

  • "connectionType": "custom.jdbc": designa una conexión a un almacén de datos de JDBC. La conexión utiliza un conector personalizado que se carga a AWS Glue Studio.

Opciones de conexión para el tipo custom.jdbc o marketplace.jdbc

  • className: cadena, obligatoria, nombre de clase de controlador.

  • connectionName: cadena, obligatoria, nombre de la conexión asociada al conector.

  • url: cadena, obligatoria, URL JDBC con marcadores de posición (${}) que se utilizan para construir la conexión al origen de datos. El marcador de posición ${secretKey} se reemplaza con el secreto del mismo nombre en AWS Secrets Manager. Consulte la documentación del almacén de datos para obtener más información sobre la construcción de la URL.

  • secretId o user/password: cadena, obligatoria, utilizada para recuperar credenciales de la URL.

  • dbTable o query: cadena, obligatoria, la tabla o consulta SQL de la que se obtienen los datos. Puede especificar dbTable o query, pero no ambos.

  • partitionColumn: cadena, opcional, el nombre de una columna entera que se utiliza para particionar. Esta opción solo funciona cuando está incluida con lowerBound, upperBound y numPartitions. Esta opción funciona de la misma manera que en el lector JDBC de Spark SQL. Para obtener más información, consulte JDBC a otras bases de datos en la Guía de Apache Spark SQL, DataFrames y conjuntos de datos.

    Los valores lowerBound y upperBound se utilizan para decidir el intervalo de partición, no para filtrar las filas de la tabla. Todas las filas de la tabla se particionan y se devuelven.

    nota

    Cuando se utiliza una consulta en lugar de un nombre de tabla, debe validar que la consulta funciona con la condición de partición especificada. Por ejemplo:

    • Si el formato de consulta es "SELECT col1 FROM table1", pruebe la consulta al agregar una cláusula WHERE al final de la consulta que utiliza la columna de partición.

    • Si su formato de consulta es “SELECT col1 FROM table1 WHERE col2=val", pruebe la consulta al ampliar la cláusula WHERE con AND y una expresión que utiliza la columna de partición.

  • lowerBound: entero, opcional, el valor mínimo de partitionColumn que se utiliza para decidir el intervalo de partición.

  • upperBound: entero, opcional, el valor máximo de partitionColumn que se utiliza para decidir el intervalo de partición.

  • numPartitions: entero, opcional, el número de particiones. Este valor, junto con lowerBound (inclusive) y upperBound (exclusivo), forma intervalos de partición para expresiones de la cláusula WHERE generadas, que se utilizan para dividir la partitionColumn.

    importante

    Preste atención al número de particiones, ya que demasiadas particiones pueden causar problemas en los sistemas de base de datos externos.

  • filterPredicate: cadena, opcional, condición adicional para filtrar datos desde el origen. Por ejemplo:

    BillingCity='Mountain View'

    Cuando se utiliza una consulta en lugar de una tabla, debe validar que la consulta funciona con el filterPredicate especificado. Por ejemplo:

    • Si el formato de consulta es "SELECT col1 FROM table1", pruebe la consulta al agregar una cláusula WHERE al final de la consulta que utiliza el predicado de filtrado.

    • Si su formato de consulta es "SELECT col1 FROM table1 WHERE col2=val", pruebe la consulta al ampliar la cláusula WHERE con AND y una expresión que utiliza el predicado de filtrado.

  • dataTypeMapping: diccionario, opcional, mapeo de tipos de datos personalizado, que crea un mapeo a partir de un tipo de datos JDBC a un tipo de datos de Glue. Por ejemplo, la opción "dataTypeMapping":{"FLOAT":"STRING"} mapea campos de datos de tipo JDBC al tipo FLOATJavaString al invocar el método ResultSet.getString() del controlador, y lo usa para construir registros de Glue. Cada controlador implementa el objeto ResultSet, por lo que el comportamiento es específico del controlador que se utiliza. Consulte la documentación del controlador JDBC para comprender cómo el controlador realiza las conversiones.

  • Los tipos de datos de AWS Glue que se admiten actualmente son:

    • FECHA

    • STRING

    • TIMESTAMP

    • INT

    • FLOAT

    • LONG

    • BIGDECIMAL

    • BYTE

    • SHORT

    • DOUBLE

    Los tipos de datos JDBC soportados son Java8 java.sql.types.

    Las asignaciones de tipos de datos predeterminados (de JDBC a AWS Glue) son:

    • DATE -> DATE

    • VARCHAR -> STRING

    • CHAR -> STRING

    • LONGNVARCHAR -> STRING

    • TIMESTAMP -> TIMESTAMP

    • INTEGER -> INT

    • FLOAT -> FLOAT

    • REAL -> FLOAT

    • BIT -> BOOLEAN

    • BOOLEAN -> BOOLEAN

    • BIGINT -> LONG

    • DECIMAL -> BIGDECIMAL

    • NUMERIC -> BIGDECIMAL

    • TINYINT -> SHORT

    • SMALLINT -> SHORT

    • DOUBLE -> DOUBLE

    Si utiliza un mapeo de tipos de datos personalizada con la opción dataTypeMapping, puede anular el mapeo de tipos de datos predeterminado. Sólo los tipos de datos JDBC enumerados en la opción dataTypeMapping se ven afectados; el mapeo predeterminado se utiliza para todos los demás tipos de datos JDBC. Puede agregar mapeos para tipos de datos JDBC adicionales si es necesario. Si un tipo de datos JDBC no está incluido en la asignación predeterminada o en una asignación personalizada, el tipo de datos se convierte al tipo de datos STRING de AWS Glue de forma predeterminada.

En los ejemplos de código Python siguientes, se muestra cómo leer desde bases de datos JDBC con controladores JDBC AWS Marketplace. Demuestra la lectura desde una base de datos y la escritura en una ubicación S3.

import sys from awsglue.transforms import * from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.job import Job ## @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) ## @type: DataSource ## @args: [connection_type = "marketplace.jdbc", connection_options = {"dataTypeMapping":{"INTEGER":"STRING"},"upperBound":"200","query":"select id, name, department from department where id < 200","numPartitions":"4", "partitionColumn":"id","lowerBound":"0","connectionName":"test-connection-jdbc"}, transformation_ctx = "DataSource0"] ## @return: DataSource0 ## @inputs: [] DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = "marketplace.jdbc", connection_options = {"dataTypeMapping":{"INTEGER":"STRING"}, "upperBound":"200","query":"select id, name, department from department where id < 200","numPartitions":"4","partitionColumn":"id","lowerBound":"0", "connectionName":"test-connection-jdbc"}, transformation_ctx = "DataSource0") ## @type: ApplyMapping ## @args: [mappings = [("department", "string", "department", "string"), ("name", "string", "name", "string"), ("id", "int", "id", "int")], transformation_ctx = "Transform0"] ## @return: Transform0 ## @inputs: [frame = DataSource0] Transform0 = ApplyMapping.apply(frame = DataSource0, mappings = [("department", "string", "department", "string"), ("name", "string", "name", "string"), ("id", "int", "id", "int")], transformation_ctx = "Transform0") ## @type: DataSink ## @args: [connection_type = "s3", format = "json", connection_options = {"path": "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0"] ## @return: DataSink0 ## @inputs: [frame = Transform0] DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Transform0, connection_type = "s3", format = "json", connection_options = {"path": "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0") job.commit()

Opciones de conexión para el tipo custom.athena o marketplace.athena

  • className: cadena, obligatoria, nombre de clase de controlador. Cuando se utiliza el conector Athena-CloudWatch, este valor de parámetro es el prefijo del nombre de clase (por ejemplo, "com.amazonaws.athena.connectors"). El conector Athena-CloudWatch se compone de dos clases: un controlador de metadatos y un controlador de registros. Si proporciona el prefijo común aquí, la API carga las clases correctas basadas en ese prefijo.

  • tableName: cadena, obligatoria, el nombre del flujo de registro de CloudWatch que se va a leer. Este fragmento de código usa el nombre de vista especial all_log_streams, lo que significa que el marco de datos dinámico devuelto contendrá datos de todos los flujos de registro incluidos en el grupo de registros.

  • schemaName: cadena, obligatoria, el nombre del grupo de registro de CloudWatch que se va a leer. Por ejemplo, /aws-glue/jobs/output.

  • connectionName: cadena, obligatoria, nombre de la conexión asociada al conector.

Para obtener opciones adicionales para este conector, consulte el archivo README (LÉAME) del conector de Amazon Athena CloudWatch en GitHub.

En el siguiente ejemplo de código de Python, se muestra cómo se lee desde un almacén de datos de Athena mediante un conector AWS Marketplace. Demuestra la lectura de Athena y la escritura en una ubicación S3.

import sys from awsglue.transforms import * from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.job import Job ## @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) ## @type: DataSource ## @args: [connection_type = "marketplace.athena", connection_options = {"tableName":"all_log_streams","schemaName":"/aws-glue/jobs/output", "connectionName":"test-connection-athena"}, transformation_ctx = "DataSource0"] ## @return: DataSource0 ## @inputs: [] DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = "marketplace.athena", connection_options = {"tableName":"all_log_streams",, "schemaName":"/aws-glue/jobs/output","connectionName": "test-connection-athena"}, transformation_ctx = "DataSource0") ## @type: ApplyMapping ## @args: [mappings = [("department", "string", "department", "string"), ("name", "string", "name", "string"), ("id", "int", "id", "int")], transformation_ctx = "Transform0"] ## @return: Transform0 ## @inputs: [frame = DataSource0] Transform0 = ApplyMapping.apply(frame = DataSource0, mappings = [("department", "string", "department", "string"), ("name", "string", "name", "string"), ("id", "int", "id", "int")], transformation_ctx = "Transform0") ## @type: DataSink ## @args: [connection_type = "s3", format = "json", connection_options = {"path": "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0"] ## @return: DataSink0 ## @inputs: [frame = Transform0] DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Transform0, connection_type = "s3", format = "json", connection_options = {"path": "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0") job.commit()

Opciones de conexión para el tipo custom.spark o marketplace.spark

  • className: cadena, obligatoria, nombre de clase de conector.

  • secretId: cadena, opcional, se utiliza para recuperar las credenciales de la conexión del conector.

  • connectionName: cadena, obligatoria, nombre de la conexión asociada al conector.

  • Otras opciones dependen del almacén de datos. Por ejemplo, las opciones de configuración de OpenSearch comienzan con el prefijo es, tal y como se describe en la documentación Elasticsearch para Apache Hadoop. Las conexiones de Spark con Snowflake utilizan opciones tales como sfUser y sfPassword, como se describe en Uso del conector de Spark en la guía Conexión a Snowflake.

En el siguiente ejemplo de código de Python, se muestra cómo se lee desde un almacén de datos de OpenSearch mediante una conexión marketplace.spark.

import sys from awsglue.transforms import * from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.job import Job ## @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) ## @type: DataSource ## @args: [connection_type = "marketplace.spark", connection_options = {"path":"test", "es.nodes.wan.only":"true","es.nodes":"https://<AWS endpoint>", "connectionName":"test-spark-es","es.port":"443"}, transformation_ctx = "DataSource0"] ## @return: DataSource0 ## @inputs: [] DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = "marketplace.spark", connection_options = {"path":"test","es.nodes.wan.only": "true","es.nodes":"https://<AWS endpoint>","connectionName": "test-spark-es","es.port":"443"}, transformation_ctx = "DataSource0") ## @type: DataSink ## @args: [connection_type = "s3", format = "json", connection_options = {"path": "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0"] ## @return: DataSink0 ## @inputs: [frame = DataSource0] DataSink0 = glueContext.write_dynamic_frame.from_options(frame = DataSource0, connection_type = "s3", format = "json", connection_options = {"path": "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0") job.commit()