API GlueContext Scala de AWS Glue

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).

def addIngestionTimeColumns

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 createDataFrameFromOptions

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 sólo con orígenes 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/"}"""))

def 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 a destinos tales como Amazon Simple Storage Service (Amazon S3), JDBC, el Catálogo de datos de AWS Glue o los flujos de datos de Apache Kafka o de Amazon Kinesis.

• connectionType — Tipo de la conexión. 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 la conexión con el receptor 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 trabajo. Se establece en un valor vacío de forma predeterminada.

Devuelve el DataSink.

def getSinkWithFormat

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.

• connectionType — Tipo de la conexión. Consulte Tipos de conexión y opciones para ETL en AWS Glue para Spark.

• options: cadena de pares de nombre-valor de JSON que proporcionan información adicional para establecer una conexión con el receptor 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 trabajo. Se establece en un valor vacío de forma predeterminada.

• format: formato de los datos que se escribirán en el destino.

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

Devuelve el DataSink.

def getSource

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

Crea una Característica DataSource que lee datos de un origen 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)
 }

def getSourceWithFormat

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 un origen como Amazon S3, JDBC o AWS Glue Data Catalog y también establece el formato de los datos almacenados en el origen.

• 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 DynamicFrame a partir de un origen 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 un DynamicFrame a partir de un origen de datos que es un 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 un DynamicFrame a partir de un origen de datos que es un MySQL usando 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 este GlueContext. Utilice este objeto SparkSession para registrar tablas y UDF con el fin de usarlas con el objeto DataFrame creado a partir de los objetos DynamicFrame.

Devuelve el objeto 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. Llama de forma interna a la API CancelTransaction de 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.