Opciones de DataFrame Conexiones personalizada y AWS Marketplace Opciones generales

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

En AWS Glue para Spark, varios métodos y transformaciones de PySpark y Scala especifican el tipo de conexión mediante 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 la página de inicio de cada tipo de conexión.

`connectionType`	Se conecta a
dynamodb	Base de datos Amazon DynamoDB
kinesis	Amazon Kinesis Data Streams
s3	Amazon S3
documentdb	Base de datos de Amazon DocumentDB (con compatibilidad con MongoDB)
opensearch	Amazon OpenSearch Service.
redshift	Base de datos de Amazon Redshift
kafka	Kafka o Amazon Managed Streaming for Apache Kafka
azurecosmos	Azure Cosmos para NoSQL.
azuresql	Azure SQL.
bigquery	Google BigQuery.
mongodb	Base de datos MongoDB, incluido MongoDB Atlas.
sqlserver	Base de datos de Microsoft SQL Server (consulte Conexiones de JDBC)
mysql	Base de datos de MySQL (consulte Conexiones de JDBC)
oracle	Base de datos de Oracle (consulte Conexiones de JDBC)
postgresql	Base de datos de PostgreSQL (consulte Conexiones de JDBC)
saphana	SAP HANA.
Snowflake	Lago de datos Snowflake
teradata	Teradata Vantage.
vertica	Vertica.
custom.*	Almacenes de datos Spark, Athena o JDBC (consulte Valores de conexiones de tipo personalizada y AWS Marketplace
marketplace.*	Almacenes de datos Spark, Athena o JDBC (consulte Valores de conexiones de tipo personalizada y AWS Marketplace)

Opciones de DataFrame para ETL en AWS Glue 5.0 para Spark

Un elemento de DataFrame ordenado en columnas con nombre es similar a una tabla y admite operaciones de estilo funcional (map/reduce/filter/etc.) y operaciones de SQL (select, project, aggregate).

Para crear un elemento de DataFrame para un origen de datos compatible con Glue, necesita lo siguiente:

conector de origen de datos ClassName
conexión de origen de datos Options

Del mismo modo, para escribir un elemento de DataFrame en un receptor de datos compatible con Glue, necesita lo mismo:

conector de receptor de datos ClassName
conexión de receptor de datos Options

Tenga en cuenta que las características de AWS Glue, como los marcadores de trabajos y las opciones de DynamicFrame como connectionName, no son compatibles con DataFrame. Para obtener más información sobre DataFrame y las operaciones compatibles, consulte la documentación de Spark sobre DataFrame.

Especificación del conector ClassName

Para especificar el origen o receptor de datos de ClassName, utilice la opción .format para proporcionar el conector correspondiente ClassName que defina el origen o receptor de datos.

Conectores de JDBC

En el caso de los conectores de JDBC, especifique jdbc como el valor de la opción .format y proporcione el controlador JDBC ClassName en la opción driver.


df = spark.read.format("jdbc").option("driver", "<DATA SOURCE JDBC DRIVER CLASSNAME>")...

df.write.format("jdbc").option("driver", "<DATA SINK JDBC DRIVER CLASSNAME>")...

En la siguiente tabla, se muestra el controlador JDBC ClassName del origen de datos compatible en AWS Glue para elementos de DataFrame.

Origen de datos	Controlador ClassName
PostgreSQL	org.postgresql.Driver
Oracle	oracle.jdbc.driver.OracleDriver
SQLServer	com.microsoft.sqlserver.jdbc.SQLServerDriver
MySQL	com.mysql.jdbc.Driver
SAPHana	com.sap.db.jdbc.Driver
Teradata	com.teradata.jdbc.TeraDriver

Conectores de Spark

Para los conectores de Spark, especifique el controlador ClassName del conector como valor de la opción .format.


df = spark.read.format("<DATA SOURCE CONNECTOR CLASSNAME>")...

df.write.format("<DATA SINK CONNECTOR CLASSNAME>")...

En la siguiente tabla, se muestra el controlador de Spark ClassName del origen de datos compatible en AWS Glue para elementos de DataFrame.

Origen de datos	ClassName
MongoDB/DocumentDB	glue.spark.mongodb
Redshift	io.github.spark_redshift_community.spark.redshift
AzureCosmos	cosmos.oltp
AzureSQL	com.microsoft.sqlserver.jdbc.spark
BigQuery	com.google.cloud.spark.bigquery
OpenSearch	org.opensearch.spark.sql
Snowflake	net.snowflake.spark.snowflake
Vertica	com.vertica.spark.datasource.VerticaSource

Especificación de la conexión Options

Para especificar la conexión Options en un origen o receptor de datos, utilice .option(<KEY>, <VALUE>) para proporcionar opciones individuales o .options(<MAP>) para proporcionar varias opciones como un mapa de valores clave.

Cada origen o receptor de datos admite su propio conjunto de conexiones Options. Para obtener detalles sobre la conexión Options disponible, consulte la documentación pública del conector de Spark del origen o receptor de datos específico que se detalla en la siguiente tabla.

Ejemplos

Los siguientes ejemplos se leen desde PostgreSQL y se escriben en SnowFlake:

Python

Ejemplo:


from awsglue.context import GlueContext
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

dataSourceClassName = "jdbc"
dataSourceOptions = {
  "driver": "org.postgresql.Driver",
  "url": "<url>",
  "user": "<user>",
  "password": "<password>",
  "dbtable": "<dbtable>",
}

dataframe = spark.read.format(className).options(**options).load()

dataSinkClassName = "net.snowflake.spark.snowflake"
dataSinkOptions = {
  "sfUrl": "<url>",
  "sfUsername": "<username>",
  "sfPassword": "<password>",
  "sfDatabase" -> "<database>",                              
  "sfSchema" -> "<schema>",                       
  "sfWarehouse" -> "<warehouse>"  
}

dataframe.write.format(dataSinkClassName).options(**dataSinkOptions).save()

Scala

Ejemplo:


import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().getOrCreate()

val dataSourceClassName = "jdbc"
val dataSourceOptions = Map(
  "driver" -> "org.postgresql.Driver",
  "url" -> "<url>",
  "user" -> "<user>",
  "password" -> "<password>",
  "dbtable" -> "<dbtable>"
)

val dataframe = spark.read.format(dataSourceClassName).options(dataSourceOptions).load()

val dataSinkClassName = "net.snowflake.spark.snowflake"
val dataSinkOptions = Map(
  "sfUrl" -> "<url>",
  "sfUsername" -> "<username>",
  "sfPassword" -> "<password>",
  "sfDatabase" -> "<database>",
  "sfSchema" -> "<schema>",
  "sfWarehouse" -> "<warehouse>"
)

dataframe.write.format(dataSinkClassName).options(dataSinkOptions).save()

Valores de conexiones de tipo personalizada y AWS Marketplace

Estos incluyen los siguientes:

"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"} asigna campos de datos de tipo FLOAT de JDBC al tipo String de Java al invocar el método ResultSet.getString() del controlador y lo utiliza para crear registros de AWS 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
- MARCA DE TIEMPO
- INT
- FLOAT
- LONG
- BIGDECIMAL
- BYTE
- SHORT
- DOBLE
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()

Opciones generales

Las opciones de esta sección se proporcionan como connection_options, pero no se aplican específicamente a un conector.

Los siguientes parámetros se utilizan generalmente al configurar los marcadores. Se pueden aplicar a los flujos de trabajo de Amazon S3 o JDBC. Para obtener más información, consulte Uso de marcadores de trabajo.

jobBookmarkKeys — Una matriz de nombres de columna.
jobBookmarkKeysSortOrder — Cadena que define cómo comparar valores en función del orden de clasificación. Valores válidos: "asc", "desc".
useS3ListImplementation — Se utiliza para administrar el rendimiento de la memoria al publicar los contenidos de los buckets de Amazon S3. Para obtener más información, consulte Cómo optimizar la gestión de la memoria en Glue AWS.

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

Características y optimizaciones

Conexiones de DynamoDB