AWS GlueAPI de Scala GlueContext - AWS Glue

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

AWS GlueAPI de Scala GlueContext

Paquete: com.amazonaws.services.glue

class GlueContext extends SQLContext(sc) ( @transient val sc : SparkContext, val defaultSourcePartitioner : PartitioningStrategy )

GlueContext es el punto de entrada para leer y escribir un DynamicFrame desde y hacia Amazon Simple Storage Service (Amazon S3), el AWS Glue Data Catalog, JDBC, etc. Esta clase ofrece funciones de utilidades para crear objetos Característica DataSource y DataSink que, a su vez, se pueden usar para leer y escribir objetos DynamicFrame.

También se puede utilizar GlueContext para establecer un número de particiones de destino (el valor predeterminado es 20) en el objeto DynamicFrame si el número de particiones creadas desde el origen es menor que un umbral mínimo para las particiones (el valor predeterminado es 10).

columnas def addIngestionTime

def addIngestionTimeColumns( df : DataFrame, timeGranularity : String = "") : dataFrame

Agrega columnas de tiempo de ingesta como ingest_year, ingest_month, ingest_day, ingest_hour, ingest_minute al DataFrame de entrada. Esta función se genera en forma automática en el script generado por AWS Glue cuando especifique una tabla del Catálogo de datos con Amazon S3 como destino. Esta función actualiza en forma automática la partición con columnas de tiempo de ingesta en la tabla de salida. Esto permite que los datos de salida se dividan automáticamente en el tiempo de ingesta sin requerir columnas de tiempo de ingesta explícitas en los datos de entrada.

  • dataFrame: el dataFrame al que anexar las columnas de tiempo de ingesta.

  • timeGranularity: la granularidad de las columnas de tiempo. Los valores válidos son “day”, “hour” y “minute”. Por ejemplo, si “hour” se transfiere a la función, el dataFrame original tendrá las columnas de tiempo “ingest_year,” “ingest_month,” “ingest_day” y “ingest_hour” anexadas.

Devuelve el marco de datos después de anexar las columnas de granularidad de tiempo.

Ejemplo:

glueContext.addIngestionTimeColumns(dataFrame, "hour")

def createDataFrame FromOptions

def createDataFrameFromOptions( connectionType : String, connectionOptions : JsonOptions, transformationContext : String = "", format : String = null, formatOptions : JsonOptions = JsonOptions.empty ) : DataSource

Muestra un DataFrame que se crea con la conexión y el formato especificados. Utilice esta función únicamente con fuentes de streaming de AWS Glue.

  • connectionType: el tipo de conexión de streaming. Los valores válidos son kinesis y kafka.

  • connectionOptions: opciones de conexión, que son diferentes para Kinesis y Kafka. Puede encontrar la lista de todas las opciones de conexión para cada origen de datos de streaming en Tipos de conexión y opciones para ETL en AWS Glue para Spark. Tenga en cuenta las siguientes diferencias en las opciones de conexión de streaming:

    • Los orígenes de streaming de Kinesis requieren streamARN, startingPosition, inferSchema y classification.

    • Los orígenes de streaming de Kafka requieren connectionName, topicName, startingOffsets, inferSchema y classification.

  • transformationContext: el contexto de transformación que se va a utilizar (opcional).

  • format: una especificación de formato (opcional). Se utiliza con una conexión de Amazon S3 o AWS Glue que admite diversos formatos. Para obtener información acerca de los formatos soportados, consulte Opciones de formato de datos para las entradas y las salidas en AWS Glue para Spark

  • formatOptions: opciones de formato para el formato especificado. Para obtener información acerca de las opciones de formatos soportados, consulte Opciones de formato de datos.

Ejemplo de origen de streaming de Amazon Kinesis:

val data_frame_datasource0 = glueContext.createDataFrameFromOptions(transformationContext = "datasource0", connectionType = "kinesis", connectionOptions = JsonOptions("""{"streamName": "example_stream", "startingPosition": "TRIM_HORIZON", "inferSchema": "true", "classification": "json"}}"""))

Ejemplo de origen de streaming de Kafka:

val data_frame_datasource0 = glueContext.createDataFrameFromOptions(transformationContext = "datasource0", connectionType = "kafka", connectionOptions = JsonOptions("""{"connectionName": "example_connection", "topicName": "example_topic", "startingPosition": "earliest", "inferSchema": "false", "classification": "json", "schema":"`column1` STRING, `column2` STRING"}"""))

forEachBatch

forEachBatch(frame, batch_function, options)

Se aplica la batch_function transferida a cada microlote que se lee desde el origen de streaming.

  • frame— El DataFrame que contiene el microlote actual.

  • batch_function: una función que se aplicará para cada microlote.

  • options: una recopilación de pares clave-valor que contiene información sobre cómo procesar microlotes. Se requieren las siguientes opciones:

    • windowSize: cantidad de tiempo que se debe dedicar al procesamiento de cada lote.

    • checkpointLocation: la ubicación donde se almacenan los puntos de verificación para el trabajo de ETL de streaming.

    • batchMaxRetries: número máximo de reintentos permitidos para este lote si se genera un error. El valor predeterminado es 3. Esta opción sólo se puede configurar para Glue versión 2.0 y superior.

Ejemplo:

glueContext.forEachBatch(data_frame_datasource0, (dataFrame: Dataset[Row], batchId: Long) => { if (dataFrame.count() > 0) { val datasource0 = DynamicFrame(glueContext.addIngestionTimeColumns(dataFrame, "hour"), glueContext) // @type: DataSink // @args: [database = "tempdb", table_name = "fromoptionsoutput", stream_batch_time = "100 seconds", // stream_checkpoint_location = "s3://from-options-testing-eu-central-1/fromOptionsOutput/checkpoint/", // transformation_ctx = "datasink1"] // @return: datasink1 // @inputs: [frame = datasource0] val options_datasink1 = JsonOptions( Map("partitionKeys" -> Seq("ingest_year", "ingest_month","ingest_day", "ingest_hour"), "enableUpdateCatalog" -> true)) val datasink1 = glueContext.getCatalogSink( database = "tempdb", tableName = "fromoptionsoutput", redshiftTmpDir = "", transformationContext = "datasink1", additionalOptions = options_datasink1).writeDynamicFrame(datasource0) } }, JsonOptions("""{"windowSize" : "100 seconds", "checkpointLocation" : "s3://from-options-testing-eu-central-1/fromOptionsOutput/checkpoint/"}"""))

- definitivamente getCatalogSink

def getCatalogSink( database : String, tableName : String, redshiftTmpDir : String = "", transformationContext : String = "" additionalOptions: JsonOptions = JsonOptions.empty, catalogId: String = null ) : DataSink

Crea un DataSink que escribe en una ubicación especificada en una tabla definida en el Data Catalog.

  • database: nombre de la base de datos en el Data Catalog.

  • tableName: nombre de la tabla en el Data Catalog.

  • redshiftTmpDir: directorio provisional que se usará con determinados receptores de datos. Se establece en un valor vacío de forma predeterminada.

  • transformationContext: contexto de transformación asociado con el receptor que utilizarán los marcadores de trabajo. Se establece en un valor vacío de forma predeterminada.

  • additionalOptions: opciones adicionales para AWS Glue.

  • catalogId: ID de catálogo (ID de cuenta) del Data Catalog al que se accede. Cuando el valor es nulo, se utiliza el ID de cuenta predeterminado del intermediario.

Devuelve el DataSink.

def getCatalogSource

def getCatalogSource( database : String, tableName : String, redshiftTmpDir : String = "", transformationContext : String = "" pushDownPredicate : String = " " additionalOptions: JsonOptions = JsonOptions.empty, catalogId: String = null ) : DataSource

Crea un objeto Característica DataSource que lee datos de una definición de tabla en el Data Catalog.

  • database: nombre de la base de datos en el Data Catalog.

  • tableName: nombre de la tabla en el Data Catalog.

  • redshiftTmpDir: directorio provisional que se usará con determinados receptores de datos. Se establece en un valor vacío de forma predeterminada.

  • transformationContext: contexto de transformación asociado con el receptor que utilizarán los marcadores de trabajo. Se establece en un valor vacío de forma predeterminada.

  • pushDownPredicate: filtra particiones sin tener que enumerar y leer todos los archivos del conjunto de datos. Para obtener más información, consulte Filtrado previo con predicados de inserción.

  • additionalOptions: conjunto de pares nombre-valor opcionales. Las opciones posibles incluyen las enumeradas en Tipos de conexión y opciones para ETL en AWS Glue para Spark, excepto por endpointUrl, streamName, bootstrap.servers, security.protocol, topicName, classification y delimiter. Otra opción soportada es catalogPartitionPredicate:

    catalogPartitionPredicate: puede transferir una expresión de catálogo para filtrar en función de las columnas de índice. Esto inserta el filtrado hacia el lado del servidor. Para obtener más información, consulte Índices de partición de AWS Glue. Tenga en cuenta que push_down_predicate y catalogPartitionPredicate utilizan sintaxis diferentes. El primero utiliza la sintaxis estándar de Spark SQL y el segundo utiliza el analizador JSQL.

  • catalogId: ID de catálogo (ID de cuenta) del Data Catalog al que se accede. Cuando el valor es nulo, se utiliza el ID de cuenta predeterminado del intermediario.

Devuelve el DataSource.

Ejemplo de origen de streaming

val data_frame_datasource0 = glueContext.getCatalogSource( database = "tempdb", tableName = "test-stream-input", redshiftTmpDir = "", transformationContext = "datasource0", additionalOptions = JsonOptions("""{ "startingPosition": "TRIM_HORIZON", "inferSchema": "false"}""") ).getDataFrame()

def getJDBCSink

def getJDBCSink( catalogConnection : String, options : JsonOptions, redshiftTmpDir : String = "", transformationContext : String = "", catalogId: String = null ) : DataSink

Crea un DataSink que escribe en una base de datos JDBC especificada en un objeto Connection en el Data Catalog. El objeto Connection tiene información para conectarse a un receptor de JDBC, incluida la URL, el nombre de usuario, la contraseña, la VPC, la subred y los grupos de seguridad.

  • catalogConnection: nombre de la conexión en el Data Catalog que contiene la URL de JDBC en la que se va a escribir.

  • options: cadena de pares de nombre-valor de JSON que proporcionan información adicional que se exige para escribir en un almacén de datos JDBC. Esto incluye:

    • dbtable (obligatorio): nombre de la tabla de JDBC. 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. El siguiente ejemplo muestra un parámetro de opciones que apunta a un esquema llamado test y a una tabla llamada test_table en la base de datos test_db.

      options = JsonOptions("""{"dbtable": "test.test_table", "database": "test_db"}""")
    • database (obligatorio): nombre de la base de datos de JDBC.

    • Todas las opciones adicionales transferidas directamente al escritor SparkSQL de JDBC. Para obtener más información, consulte el artículo acerca del origen de datos Redshift de Spark.

  • redshiftTmpDir: directorio provisional que se usará con determinados receptores de datos. Se establece en un valor vacío de forma predeterminada.

  • transformationContext: contexto de transformación asociado con el receptor que utilizarán los marcadores de trabajo. Se establece en un valor vacío de forma predeterminada.

  • catalogId: ID de catálogo (ID de cuenta) del Data Catalog al que se accede. Cuando el valor es nulo, se utiliza el ID de cuenta predeterminado del intermediario.

Código de ejemplo:

getJDBCSink(catalogConnection = "my-connection-name", options = JsonOptions("""{"dbtable": "my-jdbc-table", "database": "my-jdbc-db"}"""), redshiftTmpDir = "", transformationContext = "datasink4")

Devuelve el DataSink.

def getSink

def getSink( connectionType : String, connectionOptions : JsonOptions, transformationContext : String = "" ) : DataSink

Crea una DataSink que escribe datos en un destino como Amazon Simple Storage Service (Amazon S3), JDBC o el catálogo de datos de AWS Glue, o una transmisión de datos de Apache Kafka o Amazon Kinesis.

Devuelve el DataSink.

Formato def getSinkWith

def getSinkWithFormat( connectionType : String, options : JsonOptions, transformationContext : String = "", format : String = null, formatOptions : JsonOptions = JsonOptions.empty ) : DataSink

Crea una DataSink que escribe datos a destinos tales como Amazon S3, JDBC o el Catálogo de datos o los flujos de datos de Apache Kafka o de Amazon Kinesis. También establece el formato de los datos que se escribirán en el destino.

Devuelve el DataSink.

def getSource

def getSource( connectionType : String, connectionOptions : JsonOptions, transformationContext : String = "" pushDownPredicate ) : DataSource

Crea una Característica DataSource que lee datos de una fuente como Amazon S3, JDBC o AWS Glue Data Catalog. También soporta orígenes de datos de streaming de Kafka y Kinesis.

  • connectionType: tipo de origen de datos. Consulte Tipos de conexión y opciones para ETL en AWS Glue para Spark.

  • connectionOptions: una cadena de pares de nombre-valor de JSON que proporcionan información adicional para establecer una conexión con el origen de datos. Para obtener más información, consulte Tipos de conexión y opciones para ETL en AWS Glue para Spark.

    Un origen de streaming de Kinesis requiere las siguientes opciones de conexión: streamARN, startingPosition, inferSchema y classification.

    Un origen de streaming de Kafka requiere las siguientes opciones de conexión: connectionName, topicName, startingOffsets, inferSchema y classification.

  • transformationContext: contexto de transformación asociado con el receptor que utilizarán los marcadores de trabajo. Se establece en un valor vacío de forma predeterminada.

  • pushDownPredicate: predicado en columnas de partición.

Devuelve el DataSource.

Ejemplo de origen de streaming de Amazon Kinesis:

val kinesisOptions = jsonOptions() data_frame_datasource0 = glueContext.getSource("kinesis", kinesisOptions).getDataFrame() private def jsonOptions(): JsonOptions = { new JsonOptions( s"""{"streamARN": "arn:aws:kinesis:eu-central-1:123456789012:stream/fromOptionsStream", |"startingPosition": "TRIM_HORIZON", |"inferSchema": "true", |"classification": "json"}""".stripMargin) }

Ejemplo de origen de streaming de Kafka:

val kafkaOptions = jsonOptions() val data_frame_datasource0 = glueContext.getSource("kafka", kafkaOptions).getDataFrame() private def jsonOptions(): JsonOptions = { new JsonOptions( s"""{"connectionName": "ConfluentKafka", |"topicName": "kafka-auth-topic", |"startingOffsets": "earliest", |"inferSchema": "true", |"classification": "json"}""".stripMargin) }

Formato def getSourceWith

def getSourceWithFormat( connectionType : String, options : JsonOptions, transformationContext : String = "", format : String = null, formatOptions : JsonOptions = JsonOptions.empty ) : DataSource

Crea una Característica DataSource que lee datos de una fuente como Amazon S3, JDBC o AWS Glue Data Catalog, y también establece el formato de los datos almacenados en la fuente.

  • connectionType: tipo de origen de datos. Consulte Tipos de conexión y opciones para ETL en AWS Glue para Spark.

  • options: una cadena de pares de nombre-valor de JSON que proporciona información adicional para establecer una conexión con el origen de datos. Consulte Tipos de conexión y opciones para ETL en AWS Glue para Spark.

  • transformationContext: contexto de transformación asociado con el receptor que utilizarán los marcadores de flujo de trabajo. Se establece en un valor vacío de forma predeterminada.

  • format: formato de los datos que se almacenan en el origen. Cuando el connectionType es "s3", también puede especificar format. Puede ser "avro", "csv", "grokLog", "ion", "json", "xml", "parquet" u "orco".

  • formatOptions: una cadena de pares de nombre-valor de JSON que proporciona opciones adicionales para analizar los datos en el origen. Consulte Opciones de formato de datos.

Devuelve el DataSource.

Ejemplos

Cree un archivo DynamicFrame a partir de una fuente de datos que sea un archivo de valores separados por comas (CSV) en Amazon S3:

val datasource0 = glueContext.getSourceWithFormat( connectionType="s3", options =JsonOptions(s"""{"paths": [ "s3://csv/nycflights.csv"]}"""), transformationContext = "datasource0", format = "csv", formatOptions=JsonOptions(s"""{"withHeader":"true","separator": ","}""") ).getDynamicFrame()

Cree una DynamicFrame a partir de una fuente de datos que sea PostgreSQL mediante una conexión JDBC:

val datasource0 = glueContext.getSourceWithFormat( connectionType="postgresql", options =JsonOptions(s"""{ "url":"jdbc:postgresql://databasePostgres-1.rds.amazonaws.com:5432/testdb", "dbtable": "public.company", "redshiftTmpDir":"", "user":"username", "password":"password123" }"""), transformationContext = "datasource0").getDynamicFrame()

Cree DynamicFrame a partir de una fuente de datos que sea MySQL mediante una conexión JDBC:

val datasource0 = glueContext.getSourceWithFormat( connectionType="mysql", options =JsonOptions(s"""{ "url":"jdbc:mysql://databaseMysql-1.rds.amazonaws.com:3306/testdb", "dbtable": "athenatest_nycflights13_csv", "redshiftTmpDir":"", "user":"username", "password":"password123" }"""), transformationContext = "datasource0").getDynamicFrame()

def getSparkSession

def getSparkSession : SparkSession

Obtiene el objeto SparkSession asociado a esta instancia de GlueContext. Utilice este SparkSession objeto para registrar tablas y UDF para usarlas con las DataFrame creadas desde. DynamicFrames

Devuelve el SparkSession.

Definición de startTransaction (iniciar transacción)

def startTransaction(readOnly: Boolean):String

Inicia una nueva transacción. Llama de forma interna a la API startTransaction de Lake Formation.

  • readOnly: (booleano) indica si esta transacción debe ser de solo lectura o de lectura y escritura. Se rechazarán las escrituras realizadas con un ID de transacción de solo lectura. No es necesario confirmar las transacciones de solo lectura.

Devuelve el ID de la transacción.

Definición de commitTransaction (confirmar transacción)

def commitTransaction(transactionId: String, waitForCommit: Boolean): Boolean

Intenta confirmar la transacción especificada. Es posible que se devuelva commitTransaction antes de que la transacción haya terminado de confirmarse. Llama de forma interna a la API commitTransaction de Lake Formation.

  • transactionId: (cadena) la transacción que se confirmará.

  • waitForCommit: (booleano) determina si se devuelve commitTransaction de inmediato. El valor predeterminado es true. Si es falso, commitTransaction realiza un sondeo y espera hasta que la transacción se haya confirmado. La cantidad de tiempo de espera se limita a un minuto mediante retroceso exponencial con un máximo de seis reintentos.

Devuelve un valor booleano para indicar si se realizó o no la confirmación.

Definición de cancelTransaction (cancelar transacción)

def cancelTransaction(transactionId: String): Unit

Intenta cancelar la transacción especificada. Internamente se llama CancelTransactionAPI Lake Formation.

  • transactionId: (cadena) la transacción que se cancelará.

Devuelve una excepción de TransactionCommittedException si la transacción se había confirmado con anterioridad.

def this

def this( sc : SparkContext, minPartitions : Int, targetPartitions : Int )

Crea un objeto GlueContext con el objeto SparkContext especificado, particiones mínimas y particiones de destino.

  • sc — La SparkContext.

  • minPartitions: número mínimo de particiones.

  • targetPartitions: número de particiones de destino.

Devuelve el GlueContext.

def this

def this( sc : SparkContext )

Crear un objeto GlueContext con el SparkContext proporcionado. Establece las particiones mínimas en 10 y las particiones de destino en 20.

  • sc — La SparkContext.

Devuelve el GlueContext.

def this

def this( sparkContext : JavaSparkContext )

Crear un objeto GlueContext con el JavaSparkContext proporcionado. Establece las particiones mínimas en 10 y las particiones de destino en 20.

  • sparkContext — La JavaSparkContext.

Devuelve el GlueContext.