AWS Glue Scala GlueContext-APIs

Paket: com.amazonaws.services.glue

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

GlueContext ist der Eintrittspunkt für das Lesen und Schreiben eines DynamicFrame von und zu Amazon Simple Storage Service (Amazon S3), den AWS Glue Data Catalog, JDBC und so weiter. Diese Klasse bietet Hilfsfunktionen zum Erstellen von DataSource-Trait- und DataSink-Objekten, die wiederum zum Lesen und Schreiben von DynamicFrames verwendet werden können.

Sie können GlueContext auch verwenden, um eine Zielanzahl an Partitionen (Standard 20) im DynamicFrame festzulegen, wenn die Anzahl der von der Quelle erstellten Partitionen kleiner ist als ein Mindestwert für Partitionen (Standard 10).

def addIngestionTimeColumns

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

Hängt die Erfassungszeitspalten wie ingest_year, ingest_month, ingest_day, ingest_hour, ingest_minute an die Eingabe DataFrame an. Diese Funktion wird automatisch in dem von AWS Glue erzeugten Skript generiert, wenn Sie eine Data-Catalog-Tabelle mit Amazon S3 als Ziel angeben. Diese Funktion aktualisiert automatisch die Partition mit Erfassungszeitspalten in der Ausgabetabelle. So können die Ausgabedaten bei der Erfassung automatisch partitioniert werden, ohne dass explizite Erfassungszeitspalten in den Eingabedaten erforderlich sind.

Gibt den Datenrahmen nach dem Anhängen der Zeitgranularitätsspalten zurück.

Beispiel:

glueContext.addIngestionTimeColumns(dataFrame, "hour")

def createDataFrameFromOptions

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

Gibt einen DataFrame zurück, der mit der angegebenen Verbindung und dem Format erstellt wurde. Verwenden Sie diese Funktion nur mit AWS-Glue-Streaming-Quellen.

Beispiel für Amazon-Kinesis-Streaming-Quelle:

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

Beispiel für die Kafka-Streaming-Quelle:

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

Beispiel:

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 

Erstellt eine DataSink, die in einen Speicherort schreibt, der in einer Tabelle angegeben ist, die im Data Catalog definiert ist.

Gibt den DataSink zurück.

def getCatalogSource

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

Erstellt einen DataSource-Trait, der Daten aus einer Tabellendefinition im Data Catalog liest.

Gibt den DataSource zurück.

Beispiel für eine Streaming-Quelle

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 

Erstellt eine DataSink, die in eine JDBC-Datenbank schreibt, die im Data Catalog in einem Connection-Objekt angegeben ist. Das Connection-Objekt enthält Informationen zum Herstellen einer Verbindung zu einer JDBC-Senke, einschließlich URL, Benutzername, Passwort, VPC, Subnetz und Sicherheitsgruppen.

Beispiel-Code:

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

Gibt den DataSink zurück.

def getSink

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

Erstellt eine DataSink, die Daten in ein Ziel wie Amazon Simple Storage Service (Amazon S3), JDBC, den AWS-Glue-Datenkatalog oder einen Apache-Kafka- oder Amazon-Kinesis-Datenstrom schreibt.

Gibt den DataSink zurück.

def getSinkWithFormat

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

Erstellt eine DataSink, die Daten in ein Ziel wie Amazon S3, JDBC, den Datenkatalog oder einen Apache-Kafka- oder Amazon-Kinesis-Datenstrom schreibt. Legt auch das Format der Daten fest, die in das Ziel geschrieben werden.

Gibt den DataSink zurück.

def getSource

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

Erstellt einen DataSource-Trait, der Daten aus einer Quelle wie Amazon S3, JDBC oder dem AWS Glue Data Catalog liest. Unterstützt auch Kafka- und Kinesis-Streaming-Datenquellen.

Gibt den DataSource zurück.

Beispiel für Amazon-Kinesis-Streaming-Quelle:

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

Beispiel für die Kafka-Streaming-Quelle:

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 

Erstellt einen DataSource-Trait, der Daten aus Quellen wie Amazon S3, JDBC oder dem AWS Glue Data Catalog liest und zudem das Format für in der Quelle gespeicherte Daten festlegt.

Gibt den DataSource zurück.

Beispiele

Erstellen Sie einen DynamicFrame mit einer CSV-Datei als Datenquelle in 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()

Erstellen Sie einen DynamicFrame aus einer PostgreSQL-Datenquelle mit einer JDBC-Verbindung:

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

Erstellen Sie einen DynamicFrame aus einer MySQL-Datenquelle mit einer JDBC-Verbindung:

 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 

Ruft das SparkSession-Objekt ab, das mit diesem GlueContext verknüpft ist. Verwenden Sie dieses SparkSession-Objekt zum Registrieren von Tabellen und UDFs für die Verwendung eines von DynamicFrames erstellten DataFrame.

Gibt die SparkSession zurück.

def startTransaction

def startTransaction(readOnly: Boolean):String

Starten Sie eine neue Transaktion. Ruft intern die startTransaction-API von Lake Formation auf.

Gibt die Transaktions-ID zurück.

def commitTransaction

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

Versucht, die angegebene Transaktion zu übernehmen. commitTransaction kann zurückkehren, bevor die Transaktion abgeschlossen ist. Ruft intern die CommitTransaction-API von Lake Formation auf.

Gibt einen booleschen Wert zurück, um anzugeben, ob das Commit abgeschlossen ist oder nicht.

def cancelTransaction

def cancelTransaction(transactionId: String): Unit

Versucht, die angegebene Transaktion abzubrechen. Ruft intern die CancelTransaction-API von Lake Formation auf.

Gibt eine TransactionCommittedException-Ausnahme zurück, wenn die Transaktion zuvor festgeschrieben wurde.

def this

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

Erzeugt ein GlueContext-Objekt mit dem angegebenen SparkContext, minimalen Partitionen und Zielpartitionen.

Gibt den GlueContext zurück.

def this

def this( sc : SparkContext )

Erstellt ein GlueContext-Objekt mit dem mitgelieferten SparkContext. Legt die Mindestpartitionen auf 10 und die Zielpartitionen auf 20 fest.

Gibt den GlueContext zurück.

def this

def this( sparkContext : JavaSparkContext )

Erstellt ein GlueContext-Objekt mit dem mitgelieferten JavaSparkContext. Legt die Mindestpartitionen auf 10 und die Zielpartitionen auf 20 fest.

Gibt den GlueContext zurück.