APIs GlueContext em Scala do AWS Glue

Pacote:   com.amazonaws.services.glue

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

GlueContext é o ponto de entrada para leitura e gravação de um DynamicFrame de e para o Amazon Simple Storage Service (Amazon S3), AWS Glue Data Catalog, JDBC e assim por diante. Essa classe fornece funções de utilitário para criar objetos Característica do DataSource e DataSink que, por sua vez, podem ser usados para ler e gravar DynamicFrames.

GlueContext também pode ser usado para definir um número de partições de destino (o padrão é 20) no DynamicFrame se o número de partições criado da origem é menor do que um limite mínimo para as partições (o padrão é 10).

def addIngestionTimeColumns

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

Acrescenta colunas de tempo de ingestão, como ingest_year, ingest_month, ingest_day, ingest_hour, ingest_minute, para o DataFrame de entrada. Essa função é gerada automaticamente no script gerado pelo AWS Glue, quando você especifica uma tabela do catálogo de dados com o Amazon S3 como destino. Essa função atualiza automaticamente a partição com colunas de tempo de ingestão na tabela de saída. Isso permite que os dados de saída sejam particionados automaticamente no tempo de ingestão sem exigir colunas de tempo de ingestão explícitas nos dados de entrada.

Retorna o quadro de dados após anexar as colunas de detealhamento de tempo.

Exemplo:

glueContext.addIngestionTimeColumns(dataFrame, "hour")

def createDataFrameFromOptions

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

Retorna um DataFrame criado com a conexão e o formato especificados. Use essa função apenas com fontes de transmissão do AWS Glue.

Exemplo para fonte de transmissão do Amazon Kinesis:

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

Exemplo para fonte de transmissão do 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

Exemplo:

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 

Cria um DataSink que grava em um local especificado de uma tabela definida no Data Catalog.

Retorna o DataSink.

def getCatalogSource

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

Cria um Característica do DataSource que lê dados em uma definição de tabela do Data Catalog.

Retorna o DataSource.

Exemplo de fonte de transmissão

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 

Cria um DataSink que grava em um banco de dados JDBC especificado em um objeto Connection no Data Catalog. O objeto Connection tem informações para se conectar a um depósito JDBC, incluindo URL, nome de usuário, senha, VPC, sub-rede e grupos de segurança.

Código de exemplo:

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

Retorna o DataSink.

def getSink

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

Cria um DataSink que grava dados em um destino como Amazon Simple Storage Service (Amazon S3), JDBC, Catálogo de Dados do AWS Glue ou um fluxo de dados do Amazon Kinesis ou do Apache Kafka.

Retorna o DataSink.

def getSinkWithFormat

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

Cria um DataSink que grava dados em um destino como Amazon S3, JDBC, Catálogo de Dados ou um fluxo de dados do Amazon Kinesis ou do Apache Kafka. Também define o formato dos dados que serão gravados no destino.

Retorna o DataSink.

def getSource

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

Cria um Característica do DataSource que lê dados em uma fonte, como o Amazon S3, JDBC ou AWS Glue Data Catalog. Também oferece suporte a origens de dados de transmissão do Kafka e Kinesis.

Retorna o DataSource.

Exemplo para fonte de transmissão do 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)
}

Exemplo para fonte de transmissão do 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 

Cria um Característica do DataSource que lê dados em uma fonte, como o Amazon S3, JDBC ou AWS Glue Data Catalog, e também define o formato dos dados armazenados na fonte.

Retorna o DataSource.

Exemplos

Crie um DynamicFrame a partir de uma origem de dados que seja um arquivo com valores separados por vírgula (CSV) no 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()

Crie um DynamicFrame a partir de uma origem de dados que seja um PostgreSQL usando uma conexão 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()

Crie um DynamicFrame a partir de uma origem de dados que seja um MySQL usando uma conexão 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 

Obtém o objeto SparkSession associado a este GlueContext. Use este objeto SparkSession para registrar tabelas e UDFs a serem usados com DataFrame criado dos DynamicFrames.

Retorna o SparkSession.

def startTransaction

def startTransaction(readOnly: Boolean):String

Iniciar uma nova transação. Chama internamente a API startTransaction do Lake Formation.

Retorna o ID da transação.

def commitTransaction

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

Tenta confirmar a transação especificada. commitTransaction pode retornar antes que a transação tenha terminado de confirmar. Chama internamente a API commitTransaction do Lake Formation.

Retorna um booleano para indicar se a confirmação foi feita ou não.

def cancelTransaction

def cancelTransaction(transactionId: String): Unit

Tenta cancelar a transação especificada. Chama internamente a API cancelTransaction do Lake Formation.

Retorna uma exceção TransactionCommittedException se a transação tiver sido confirmada anteriormente.

def this

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

Cria um objeto GlueContext usando o SparkContext especificado, um mínimo de partições e partições de destino.

Retorna o GlueContext.

def this

def this( sc : SparkContext )

Cria um objeto GlueContext com o SparkContext fornecido. Define o mínimo de partições como 10 e as partições de destino como 20.

Retorna o GlueContext.

def this

def this( sparkContext : JavaSparkContext )

Cria um objeto GlueContext com o JavaSparkContext fornecido. Define o mínimo de partições como 10 e as partições de destino como 20.

Retorna o GlueContext.