本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

AWS Glue Scala GlueContext API

Package: com.amazonaws.services.glue

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

GlueContext 是讀取和寫入 DynamicFrame 至 Amazon Simple Storage Service (Amazon S3)、 AWS Glue Data Catalog 、JDBC 等的進入點。此類別提供公用程式函數來建立 DataSource 特徵 和 DataSink 物件，從而用於讀取和寫入 DynamicFrame。

如果從來源建立的分割區數低於分割區的閾值下限 (預設 10)，您也可以使用 GlueContext 來設定在 DynamicFrame 中的分割區目標數 (預設 20)。

def addIngestionTimeColumns

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

附加擷取時間欄 (如 ingest_year、ingest_month、ingest_day、ingest_hour、ingest_minute) 到輸入 DataFrame。當您指定以 Amazon S3 為目標的 Data Catalog 資料表時，此函數會在 AWS Glue 產生的指令碼中自動產生。此函數會自動使用輸出資料表上的擷取時間欄來更新分割區。這可讓輸出資料在擷取時間自動分割，而不需要輸入資料中的明確擷取時間欄。

傳回附加時間粒度欄後的資料框架。

範例：

glueContext.addIngestionTimeColumns(dataFrame, "hour")

def createDataFrameFromOptions

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

傳回使用指定的連線和格式建立的 DataFrame。此函數僅適用於 Glue AWS 串流來源。

Amazon Kinesis 串流來源範例：

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

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

範例：

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 

建立 DataSink，以便寫入 Data Catalog 中定義之資料表中指定的位置。

傳回 DataSink。

def getCatalogSource

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

建立 DataSource 特徵，以便從 Data Catalog 中的資料表定義中讀取資料。

傳回 DataSource。

串流來源範例

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 

建立 DataSink，以便寫入 Data Catalog 中 Connection 物件所指定的 JDBC 資料庫。此 Connection 物件擁有用來對 JDBC 目的地連線的資訊 (包括 URL、使用者名稱、密碼、VPC、子網路和安全群組)。

範例程式碼：

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

傳回 DataSink。

def getSink

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

建立 DataSink，將資料寫入 Amazon Simple Storage Service (Amazon S3)、JDBC 或 AWS Glue Data Catalog 等目的地，或 Apache Kafka 或 Amazon Kinesis 資料串流。

傳回 DataSink。

def getSinkWithFormat

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

建立 DataSink，將資料寫入 Amazon S3、JDBC 或 Data Catalog 等目的地，或 Apache Kafka 或 Amazon Kinesis 資料串流。也會設定要寫入目的地的資料格式。

傳回 DataSink。

def getSource

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

建立從 Amazon S3、JDBC 或 Glue Data Catalog AWS 等來源DataSource 特徵讀取資料的 。也支援 Kafka 和 Kinesis 串流資料來源。

傳回 DataSource。

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

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 

建立從 Amazon S3、JDBC 或 AWS Glue Data Catalog 等來源DataSource 特徵讀取資料的 ，並設定存放在來源中的資料格式。

傳回 DataSource。

範例

從 Amazon S3 上逗號分隔值 (CSV) 檔案的資料來源建立 DynamicFrame：

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

從使用 JDBC 連線的 PostgreSQL 資料來源建立 DynamicFrame：

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

從使用 JDBC 連線的 MySQL 資料來源建立 DynamicFrame：

 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 

取得與此 GlueContext 相關聯的 SparkSession 物件。使用此 SparkSession 物件以將資料表與 UDF 註冊為與從 DynamicFrames 建立的 DataFrame 搭配使用。

傳回 SparkSession。

def startTransaction

def startTransaction(readOnly: Boolean):String

開始新交易。內部呼叫 Lake Formation startTransaction API。

傳回交易 ID。

def commitTransaction

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

嘗試遞交指定的交易。commitTransaction 可能會在交易完成遞交之前返回。內部呼叫 Lake Formation commitTransaction API。

傳回一個布林值，指示遞交是否完成。

def cancelTransaction

def cancelTransaction(transactionId: String): Unit

嘗試取消指定的交易。內部呼叫 Lake Formation CancelTransaction API。

如果交易先前已遞交，傳回 TransactionCommittedException 例外狀況。

def 此

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

使用指定的 SparkContext、最小分割區和分割區目標來建立 GlueContext 物件。

傳回 GlueContext。

def 此

def this( sc : SparkContext )

透過提供的 SparkContext 建立 GlueContext 物件。將分割區的最小值設為 10，目標分割區設為 20。

傳回 GlueContext。

def 此

def this( sparkContext : JavaSparkContext )

透過提供的 JavaSparkContext 建立 GlueContext 物件。將分割區的最小值設為 10，目標分割區設為 20。

傳回 GlueContext。