Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

AWS Glue GlueContext API Scala

Pacchetto: com.amazonaws.services.glue

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

GlueContext è il punto di ingresso per la lettura e la scrittura di un DynamicFrame da e verso Amazon Simple Storage Service (Amazon S3), il catalogo dati AWS Glue, JDBC e così via. Questa classe fornisce funzioni di utilità per creare oggetti Proprietà DataSource e DataSink che possono in cambio essere utilizzati per leggere e scrivere DynamicFrame.

GlueContext può anche essere usato per impostare un numero target di partizioni (per impostazione predefinita 20) nel DynamicFrame se il numero di partizioni create dalla sorgente è inferiore alla soglia minima delle partizioni (per impostazione predefinita 10).

def Colonne addIngestionTime

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

Aggiunge colonne del tempo di importazione dati come ingest_year, ingest_month, ingest_day, ingest_hour, ingest_minute al DataFrame di input. Questa funzione viene generata automaticamente nello script generato da AWS Glue quando si specifica una tabella del catalogo dati con Amazon S3 come destinazione. Questa funzione aggiorna automaticamente la partizione con le colonne del tempo di importazione dati nella tabella di output. Ciò consente ai dati di output di venire partizionati automaticamente nel tempo di importazione dati senza necessitare di colonne di tempo di inserimento esplicite nei dati di input.

Restituisce il frame di dati dopo l'aggiunta di colonne di granularità di tempo.

Esempio:

glueContext.addIngestionTimeColumns(dataFrame, "hour")

def createDataFrame FromOptions

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

Restituisce un DataFrame creato con la connessione e il formato specificati. Usa questa funzione solo con le sorgenti di streaming AWS Glue.

Esempio per l'origine di streaming Amazon Kinesis:

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

Esempio per l'origine di streaming 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

Esempio:

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 che scrive in una posizione specificata di una tabella definita nel catalogo dati.

Restituisce il 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 Proprietà DataSource che legge dati da una definizione di tabella nel catalogo di dati.

Restituisce il DataSource.

Esempio di origine di 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 oggetto DataSink che scrive in un database JDBC specificato in un oggetto Connection nel catalogo dati. L'oggetto Connection dispone di informazioni per connettersi a un sink JDBC, compresi URL, nome utente, password, VPC, sottorete e gruppi di sicurezza.

Codice di esempio:

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

Restituisce il DataSink.

def getSink

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

Crea un file DataSink che scrive dati su una destinazione come Amazon Simple Storage Service (Amazon S3), JDBC o Glue Data Catalog o AWS un flusso di dati Apache Kafka o Amazon Kinesis.

Restituisce il DataSink.

Formato def getSinkWith

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

Crea un file DataSink che scrive dati su una destinazione come Amazon S3, JDBC o Data Catalog o un flusso di dati Apache Kafka o Amazon Kinesis. Imposta anche il formato per i dati da scrivere nella destinazione.

Restituisce il DataSink.

def getSource

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

Crea un file Proprietà DataSource che legge i dati da una fonte come Amazon S3, JDBC o Glue Data Catalog. AWS Supporta anche origini dati di streaming Kafka e Kinesis.

Restituisce il DataSource.

Esempio per l'origine di streaming 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)
}

Esempio per l'origine di streaming 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 un file Proprietà DataSource che legge i dati da una fonte come Amazon S3, JDBC o AWS Glue Data Catalog e imposta anche il formato dei dati archiviati nell'origine.

Restituisce il DataSource.

Examples (Esempi)

Crea un file DynamicFrame da un'origine dati che sia un file con valori separati da virgole (CSV) su 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()

Crea un file DynamicFrame da un'origine dati che sia PostgreSQL utilizzando una connessione 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()

Crea un file DynamicFrame da un'origine dati che è MySQL utilizzando una connessione 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 

Ottiene l'oggetto SparkSession associato a questo GlueContext. Utilizzate questo SparkSession oggetto per registrare tabelle e UDF da utilizzare con DataFrame created from. DynamicFrames

Restituisce il. SparkSession

def startTransaction

def startTransaction(readOnly: Boolean):String

Avvia una nuova transazione. Chiama internamente l'API Lake Formation startTransaction.

Restituisce l'ID transazione.

def commitTransaction

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

Tenta di eseguire il commit della transazione specificata. commitTransaction può restituire prima che la transazione abbia terminato il commit. Chiama internamente l'API Lake Formation commitTransaction.

Restituisce un valore booleano per indicare se il commit sia stato eseguito o meno.

def cancelTransaction

def cancelTransaction(transactionId: String): Unit

Tenta di annullare la transazione specificata. Richiama internamente l'CancelTransactionAPI Lake Formation.

Restituisce un'eccezione TransactionCommittedException se è stato precedentemente eseguito il commit della transazione.

def this

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

Crea un oggetto GlueContext utilizzando lo SparkContext specificato, le partizioni minime e quelle target.

Restituisce il GlueContext.

def this

def this( sc : SparkContext )

Crea un oggetto GlueContext con il SparkContext fornito. Imposta il numero minimo di partizioni a 10 e a le partizioni target a 20.

Restituisce il GlueContext.

def this

def this( sparkContext : JavaSparkContext )

Crea un oggetto GlueContext con il JavaSparkContext fornito. Imposta il numero minimo di partizioni a 10 e a le partizioni target a 20.

Restituisce il GlueContext.